Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b7577c4552 | ||
|
|
43b634a135 |
+21
-1
@@ -8,6 +8,24 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
|
|||||||
|
|
||||||
## [Unreleased]
|
## [Unreleased]
|
||||||
|
|
||||||
|
## [0.9.0] – 2026-10-15
|
||||||
|
|
||||||
|
### Hinzugefügt
|
||||||
|
- **Regeln kopieren** (Knopf „Kopieren“): alle Einstellungen inklusive Archiv
|
||||||
|
übernommen, Name „… (Kopie)“. Die Kopie ist zunächst **deaktiviert**, damit sie nicht
|
||||||
|
mit dem Original um dieselben Dateien konkurriert; Protokoll und Buchführung
|
||||||
|
beginnen neu.
|
||||||
|
|
||||||
|
## [0.8.0] – 2026-10-12
|
||||||
|
|
||||||
|
### Hinzugefügt
|
||||||
|
- **Anbindungen kopieren** (Knopf „Kopieren“): Formular vorausgefüllt, Name
|
||||||
|
„… (Kopie)“. Eigene Zugangsdaten der Vorlage werden übernommen (eigenes Exemplar),
|
||||||
|
solange kein neues Passwort eingegeben wird; ein eigener SSH-Schlüssel wird
|
||||||
|
weiterverwendet; gespeicherte Zugangsdaten bleiben ausgewählt. „Testen“ im
|
||||||
|
Kopie-Formular nutzt die übernommenen Zugangsdaten. Der SFTP-Hostschlüssel wird nur
|
||||||
|
bei gleichem Server und Port übernommen.
|
||||||
|
|
||||||
## [0.7.0] – 2026-10-12
|
## [0.7.0] – 2026-10-12
|
||||||
|
|
||||||
### Hinzugefügt
|
### Hinzugefügt
|
||||||
@@ -259,7 +277,9 @@ Erste Version.
|
|||||||
- Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
|
- Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
|
||||||
Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.
|
Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.
|
||||||
|
|
||||||
[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.7.0...HEAD
|
[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.9.0...HEAD
|
||||||
|
[0.9.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.8.0...v0.9.0
|
||||||
|
[0.8.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.7.0...v0.8.0
|
||||||
[0.7.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.6.0...v0.7.0
|
[0.7.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.6.0...v0.7.0
|
||||||
[0.6.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.5.0...v0.6.0
|
[0.6.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.5.0...v0.6.0
|
||||||
[0.5.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.4.0...v0.5.0
|
[0.5.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.4.0...v0.5.0
|
||||||
|
|||||||
@@ -21,10 +21,37 @@ Zeiten in Ortszeit (Europe/Berlin).
|
|||||||
| 2026-10-11 | 00:54 | 01:16 | 0:22 | – | SFTP-Quelle, Zugangsdaten, Anmeldung (0.5.0) |
|
| 2026-10-11 | 00:54 | 01:16 | 0:22 | – | SFTP-Quelle, Zugangsdaten, Anmeldung (0.5.0) |
|
||||||
| 2026-10-11 | 08:43 | 08:55 | 0:12 | – | Anmeldung je Netz (0.6.0) |
|
| 2026-10-11 | 08:43 | 08:55 | 0:12 | – | Anmeldung je Netz (0.6.0) |
|
||||||
| 2026-10-12 | 00:14 | 00:34 | 0:20 | – | Archiv je Regel (0.7.0) |
|
| 2026-10-12 | 00:14 | 00:34 | 0:20 | – | Archiv je Regel (0.7.0) |
|
||||||
| **Summe** | | | **3:49** | | |
|
| 2026-10-12 | 00:46 | 00:55 | 0:09 | – | Anbindungen kopieren (0.8.0) |
|
||||||
|
| 2026-10-15 | 09:17 | 09:26 | 0:09 | – | Regeln kopieren (0.9.0) |
|
||||||
|
| **Summe** | | | **4:07** | | |
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
## 2026-10-15 · 09:17–09:26 · Regeln kopieren (0.9.0)
|
||||||
|
|
||||||
|
**Erledigt** – Knopf „Kopieren“, Route `/rules/{id}/copy` (GET Formular, POST
|
||||||
|
speichern), `rule_copy_name`. 5 Tests (362 gesamt); Gegenprobe „Kopie aktiv“ schlägt
|
||||||
|
an (2 Tests).
|
||||||
|
|
||||||
|
**Entscheidung** – Kopie startet deaktiviert: Mit identischer Quelle würden Original
|
||||||
|
und Kopie sonst sofort parallel dieselben Dateien bearbeiten (doppelte Lieferung bzw.
|
||||||
|
Kollisionen). Protokoll und Buchführung (`copied_files`, `archived_files`) werden
|
||||||
|
bewusst nicht mitkopiert – die Kopie ist eine neue Regel.
|
||||||
|
|
||||||
|
## 2026-10-12 · 00:46–00:55 · Anbindungen kopieren (0.8.0)
|
||||||
|
|
||||||
|
**Erledigt** – Knopf „Kopieren“, Route `/mounts/{id}/copy` (GET Formular, POST
|
||||||
|
speichern), `save_mount(copy_from=…)`, Formular-Test mit Vorlage. 7 Tests (357 gesamt);
|
||||||
|
Gegenprobe ohne Übernahme der Zugangsdaten: 3 Tests schlagen an.
|
||||||
|
|
||||||
|
**Entscheidungen**
|
||||||
|
- Eigene Zugangsdaten werden als eigenes Exemplar kopiert (nicht geteilt) – eigene
|
||||||
|
Zugangsdaten verschwinden mit ihrer Anbindung, eine geteilte Nutzung wäre
|
||||||
|
dafür der Fall „Zugangsdaten“.
|
||||||
|
- Eigenes Schlüsselpaar bleibt gleich (schon in `authorized_keys`).
|
||||||
|
- Hostschlüssel nur bei gleichem Server/Port übernehmen, sonst würde der Test an
|
||||||
|
einem anderen Server mit „Hostschlüssel geändert“ scheitern.
|
||||||
|
|
||||||
## 2026-10-12 · 00:14–00:34 · Archiv je Regel (0.7.0)
|
## 2026-10-12 · 00:14–00:34 · Archiv je Regel (0.7.0)
|
||||||
|
|
||||||
**Erledigt**
|
**Erledigt**
|
||||||
|
|||||||
@@ -28,6 +28,8 @@ Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht
|
|||||||
|
|
||||||
| Version | Thema |
|
| Version | Thema |
|
||||||
|---|---|
|
|---|---|
|
||||||
|
| 0.9.0 | Regeln kopieren (Kopie zunächst deaktiviert) |
|
||||||
|
| 0.8.0 | Anbindungen kopieren inklusive Zugangsdaten |
|
||||||
| 0.7.0 | Archiv je Regel: Kopie ins Ziel und ins Archiv, Quelle erst danach löschen |
|
| 0.7.0 | Archiv je Regel: Kopie ins Ziel und ins Archiv, Quelle erst danach löschen |
|
||||||
| 0.6.0 | Anmeldung je Netz: ohne Passwort / mit Passwort / abgewiesen, Schutz vor Aussperren |
|
| 0.6.0 | Anmeldung je Netz: ohne Passwort / mit Passwort / abgewiesen, Schutz vor Aussperren |
|
||||||
| 0.5.0 | SFTP als Quelle; gespeicherte Zugangsdaten für SMB/SFTP-Passwörter (Seite „Zugangsdaten“); Anmeldung abschaltbar (vertrauenswürdige Netze / ganz) |
|
| 0.5.0 | SFTP als Quelle; gespeicherte Zugangsdaten für SMB/SFTP-Passwörter (Seite „Zugangsdaten“); Anmeldung abschaltbar (vertrauenswürdige Netze / ganz) |
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@
|
|||||||
services:
|
services:
|
||||||
netmover:
|
netmover:
|
||||||
build: .
|
build: .
|
||||||
image: netmover:0.7.0
|
image: netmover:0.9.0
|
||||||
container_name: netmover
|
container_name: netmover
|
||||||
restart: unless-stopped
|
restart: unless-stopped
|
||||||
cap_add:
|
cap_add:
|
||||||
|
|||||||
+30
-1
@@ -72,6 +72,22 @@ Stabilitätsfenster dort mindestens 2 Sekunden betragen.
|
|||||||
| **Beim Start automatisch einhängen** | Nach einem Neustart des Servers wieder einhängen. Fällt das NAS aus, versucht NetMover es außerdem jede Minute erneut. |
|
| **Beim Start automatisch einhängen** | Nach einem Neustart des Servers wieder einhängen. Fällt das NAS aus, versucht NetMover es außerdem jede Minute erneut. |
|
||||||
| **Aktiv** | Deaktivierte Anbindungen werden von Regeln nicht benutzt. |
|
| **Aktiv** | Deaktivierte Anbindungen werden von Regeln nicht benutzt. |
|
||||||
|
|
||||||
|
### Kopieren {#anbindung-kopieren}
|
||||||
|
|
||||||
|
Für eine ähnliche Anbindung – etwa eine zweite Freigabe auf demselben NAS – bei der
|
||||||
|
Vorlage auf **Kopieren** klicken. Das Formular ist mit allen Werten vorausgefüllt
|
||||||
|
(Name „… (Kopie)“); meist muss nur die Freigabe geändert werden.
|
||||||
|
|
||||||
|
- **Zugangsdaten werden übernommen**, solange das Passwortfeld leer bleibt. Die Kopie
|
||||||
|
bekommt ein eigenes Exemplar – das Löschen der Vorlage betrifft sie nicht.
|
||||||
|
Gespeicherte Zugangsdaten sind einfach vorausgewählt.
|
||||||
|
- Ein **eigener SSH-Schlüssel** der Vorlage wird weiter benutzt; ein neuer Eintrag in
|
||||||
|
`authorized_keys` ist nicht nötig.
|
||||||
|
- **Testen** funktioniert im Kopie-Formular bereits mit den übernommenen Zugangsdaten.
|
||||||
|
|
||||||
|
Tipp: Nutzen mehrere Anbindungen dieselben Zugangsdaten, lohnt sich ein Eintrag unter
|
||||||
|
[Zugangsdaten](#zugangsdaten) – dann gilt ein neues Passwort sofort überall.
|
||||||
|
|
||||||
### Testen {#anbindung-testen}
|
### Testen {#anbindung-testen}
|
||||||
|
|
||||||
**Testen** prüft drei Dinge einzeln und sagt im Klartext, was nicht klappt:
|
**Testen** prüft drei Dinge einzeln und sagt im Klartext, was nicht klappt:
|
||||||
@@ -210,6 +226,17 @@ Synology-Verwaltungsordner (`@eaDir`, `#recycle`).
|
|||||||
| **Modus setzen** | z. B. `0644`; leer = wie die Quelldatei |
|
| **Modus setzen** | z. B. `0644`; leer = wie die Quelldatei |
|
||||||
| **Quarantäneordner** | Dateien, die mehrmals hintereinander scheitern, werden dorthin verschoben (relativ zur Quell-Anbindung), statt es endlos zu versuchen. Leer = keine Quarantäne. |
|
| **Quarantäneordner** | Dateien, die mehrmals hintereinander scheitern, werden dorthin verschoben (relativ zur Quell-Anbindung), statt es endlos zu versuchen. Leer = keine Quarantäne. |
|
||||||
|
|
||||||
|
### Regel kopieren {#regel-kopieren}
|
||||||
|
|
||||||
|
Für eine ähnliche Regel – etwa für einen zweiten Quell-Unterordner – bei der Vorlage
|
||||||
|
auf **Kopieren** klicken. Alle Einstellungen (Filter, Ziel, Archiv, Sicherheit,
|
||||||
|
Rechte) sind übernommen, der Name lautet „… (Kopie)“.
|
||||||
|
|
||||||
|
Die Kopie ist zunächst **deaktiviert**: Mit gleicher Quelle würde sie sonst sofort mit
|
||||||
|
dem Original um dieselben Dateien konkurrieren. Also Quelle bzw. Ziel anpassen,
|
||||||
|
speichern, **Probelauf**, dann **Aktivieren**. Protokoll und Buchführung beginnen bei
|
||||||
|
der Kopie neu.
|
||||||
|
|
||||||
### Probelauf {#probelauf}
|
### Probelauf {#probelauf}
|
||||||
|
|
||||||
**Probelauf** zeigt in einer Tabelle, welche Datei wohin ginge – **ohne etwas zu
|
**Probelauf** zeigt in einer Tabelle, welche Datei wohin ginge – **ohne etwas zu
|
||||||
@@ -412,7 +439,9 @@ deaktiviert, Zugangsdaten müssen neu eingegeben werden.
|
|||||||
### Jede Person in einen eigenen Ordner {#beispiel-unterordner}
|
### Jede Person in einen eigenen Ordner {#beispiel-unterordner}
|
||||||
|
|
||||||
Scannt jede Person in einen eigenen Unterordner (`Scans/max`, `Scans/erika`): pro
|
Scannt jede Person in einen eigenen Unterordner (`Scans/max`, `Scans/erika`): pro
|
||||||
Person eine Regel mit eigenem **Unterordner** in der Quelle und eigenem Ziel.
|
Person eine Regel mit eigenem **Unterordner** in der Quelle und eigenem Ziel. Am
|
||||||
|
schnellsten: die erste Regel einrichten, dann **Kopieren** und nur Unterordner und
|
||||||
|
Ziel ändern.
|
||||||
|
|
||||||
### Kopie als zusätzliche Sicherung {#beispiel-kopie}
|
### Kopie als zusätzliche Sicherung {#beispiel-kopie}
|
||||||
|
|
||||||
|
|||||||
@@ -1,3 +1,3 @@
|
|||||||
"""NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
|
"""NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
|
||||||
|
|
||||||
__version__ = "0.7.0"
|
__version__ = "0.9.0"
|
||||||
|
|||||||
+48
-5
@@ -294,7 +294,11 @@ def save_mount(
|
|||||||
mounts: MountManager,
|
mounts: MountManager,
|
||||||
data: MountInput,
|
data: MountInput,
|
||||||
mount_id: int | None = None,
|
mount_id: int | None = None,
|
||||||
|
copy_from: Mount | None = None,
|
||||||
) -> tuple[Mount | None, Errors]:
|
) -> tuple[Mount | None, Errors]:
|
||||||
|
"""``copy_from``: Vorlage beim Kopieren – eigene Zugangsdaten (Passwort bzw.
|
||||||
|
Schlüsselpaar) werden übernommen, solange kein neues Passwort eingegeben ist;
|
||||||
|
die Kopie bekommt ein eigenes Exemplar."""
|
||||||
errors: Errors = {}
|
errors: Errors = {}
|
||||||
m = s.get(Mount, mount_id) if mount_id is not None else None
|
m = s.get(Mount, mount_id) if mount_id is not None else None
|
||||||
if mount_id is not None and m is None:
|
if mount_id is not None and m is None:
|
||||||
@@ -349,6 +353,14 @@ def save_mount(
|
|||||||
setattr(m, f, getattr(data, f))
|
setattr(m, f, getattr(data, f))
|
||||||
if not data.type.is_kernel_mount:
|
if not data.type.is_kernel_mount:
|
||||||
m.auto_mount = False
|
m.auto_mount = False
|
||||||
|
if (
|
||||||
|
mount_id is None
|
||||||
|
and copy_from is not None
|
||||||
|
and data.type is MountType.SFTP
|
||||||
|
and copy_from.type is MountType.SFTP
|
||||||
|
and (copy_from.server, copy_from.port) == (data.server, data.port)
|
||||||
|
):
|
||||||
|
m.host_key = copy_from.host_key # gleicher Server – Hostschlüssel bleibt gültig
|
||||||
|
|
||||||
# Zugangsdaten: entweder ein gespeicherter Schlüssel aus der Schlüsselverwaltung
|
# Zugangsdaten: entweder ein gespeicherter Schlüssel aus der Schlüsselverwaltung
|
||||||
# (geteilt, wird hier nie verändert) oder eigene Zugangsdaten nur dieser Anbindung.
|
# (geteilt, wird hier nie verändert) oder eigene Zugangsdaten nur dieser Anbindung.
|
||||||
@@ -386,21 +398,32 @@ def save_mount(
|
|||||||
else:
|
else:
|
||||||
m.username = None
|
m.username = None
|
||||||
cred = own
|
cred = own
|
||||||
|
template = (
|
||||||
|
copy_from.credential
|
||||||
|
if copy_from is not None
|
||||||
|
and copy_from.credential is not None
|
||||||
|
and not copy_from.credential.managed
|
||||||
|
and copy_from.credential.kind is data.auth
|
||||||
|
else None
|
||||||
|
)
|
||||||
if cred is None or cred.kind is not data.auth:
|
if cred is None or cred.kind is not data.auth:
|
||||||
if data.auth is CredentialKind.SSH_KEY:
|
if template is not None and (data.auth is CredentialKind.SSH_KEY or not data.password):
|
||||||
private, public = generate_ssh_keypair(f"netmover@{m.slug}")
|
# Kopie: dasselbe Passwort bzw. Schlüsselpaar (steht schon in authorized_keys)
|
||||||
secret, pub = private, public
|
secret_enc, pub = template.secret_enc, template.public_key
|
||||||
|
elif data.auth is CredentialKind.SSH_KEY:
|
||||||
|
private, pub = generate_ssh_keypair(f"netmover@{m.slug}")
|
||||||
|
secret_enc = secrets.encrypt(private)
|
||||||
else:
|
else:
|
||||||
if not data.password:
|
if not data.password:
|
||||||
errors["password"] = "Passwort fehlt." # noqa: S105
|
errors["password"] = "Passwort fehlt." # noqa: S105
|
||||||
s.rollback()
|
s.rollback()
|
||||||
return None, errors
|
return None, errors
|
||||||
secret, pub = data.password, None
|
secret_enc, pub = secrets.encrypt(data.password), None
|
||||||
if cred is None:
|
if cred is None:
|
||||||
cred = Credential(name=f"mount:{m.slug}", managed=False)
|
cred = Credential(name=f"mount:{m.slug}", managed=False)
|
||||||
s.add(cred)
|
s.add(cred)
|
||||||
cred.kind = data.auth
|
cred.kind = data.auth
|
||||||
cred.secret_enc = secrets.encrypt(secret)
|
cred.secret_enc = secret_enc
|
||||||
cred.public_key = pub
|
cred.public_key = pub
|
||||||
elif data.password and data.auth is CredentialKind.PASSWORD:
|
elif data.password and data.auth is CredentialKind.PASSWORD:
|
||||||
cred.secret_enc = secrets.encrypt(data.password)
|
cred.secret_enc = secrets.encrypt(data.password)
|
||||||
@@ -421,6 +444,16 @@ def save_mount(
|
|||||||
return m, errors
|
return m, errors
|
||||||
|
|
||||||
|
|
||||||
|
def copy_name(s: Session, name: str) -> str:
|
||||||
|
"""„NAS (Kopie)“, „NAS (Kopie 2)“, … – frei in der Datenbank."""
|
||||||
|
base = f"{name} (Kopie)"[:100]
|
||||||
|
candidate, n = base, 1
|
||||||
|
while s.scalar(select(Mount.id).where(Mount.name == candidate)) is not None:
|
||||||
|
n += 1
|
||||||
|
candidate = f"{name} (Kopie {n})"[:100]
|
||||||
|
return candidate
|
||||||
|
|
||||||
|
|
||||||
def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None:
|
def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None:
|
||||||
m = s.get(Mount, mount_id)
|
m = s.get(Mount, mount_id)
|
||||||
if m is None:
|
if m is None:
|
||||||
@@ -635,6 +668,16 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
|
|||||||
return d, errors
|
return d, errors
|
||||||
|
|
||||||
|
|
||||||
|
def rule_copy_name(s: Session, name: str) -> str:
|
||||||
|
"""„Scans (Kopie)“, „Scans (Kopie 2)“, … – frei in der Datenbank."""
|
||||||
|
base = f"{name} (Kopie)"[:100]
|
||||||
|
candidate, n = base, 1
|
||||||
|
while s.scalar(select(Rule.id).where(Rule.name == candidate)) is not None:
|
||||||
|
n += 1
|
||||||
|
candidate = f"{name} (Kopie {n})"[:100]
|
||||||
|
return candidate
|
||||||
|
|
||||||
|
|
||||||
def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rule | None, Errors]:
|
def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rule | None, Errors]:
|
||||||
dup = s.scalar(select(Rule.id).where(Rule.name == d.name))
|
dup = s.scalar(select(Rule.id).where(Rule.name == d.name))
|
||||||
if dup is not None and dup != rule_id:
|
if dup is not None and dup != rule_id:
|
||||||
|
|||||||
+67
-13
@@ -38,11 +38,13 @@ from netmover.admin import (
|
|||||||
AdminError,
|
AdminError,
|
||||||
MountInput,
|
MountInput,
|
||||||
RuleInput,
|
RuleInput,
|
||||||
|
copy_name,
|
||||||
credential_for_test,
|
credential_for_test,
|
||||||
delete_mount,
|
delete_mount,
|
||||||
mount_input_from_model,
|
mount_input_from_model,
|
||||||
parse_mount_form,
|
parse_mount_form,
|
||||||
parse_rule_form,
|
parse_rule_form,
|
||||||
|
rule_copy_name,
|
||||||
rule_input_from_model,
|
rule_input_from_model,
|
||||||
save_mount,
|
save_mount,
|
||||||
save_rule,
|
save_rule,
|
||||||
@@ -591,10 +593,12 @@ def _mount_form(
|
|||||||
errors: dict[str, str],
|
errors: dict[str, str],
|
||||||
mount: Mount | None = None,
|
mount: Mount | None = None,
|
||||||
status_code: int = 200,
|
status_code: int = 200,
|
||||||
|
copy_from: Mount | None = None,
|
||||||
) -> Response:
|
) -> Response:
|
||||||
|
owner = mount or copy_from
|
||||||
pub = (
|
pub = (
|
||||||
mount.credential.public_key
|
owner.credential.public_key
|
||||||
if mount is not None and mount.credential and not mount.credential.managed
|
if owner is not None and owner.credential and not owner.credential.managed
|
||||||
else None
|
else None
|
||||||
)
|
)
|
||||||
return render(
|
return render(
|
||||||
@@ -607,8 +611,9 @@ def _mount_form(
|
|||||||
mount=mount,
|
mount=mount,
|
||||||
public_key=pub,
|
public_key=pub,
|
||||||
has_secret=bool(
|
has_secret=bool(
|
||||||
mount is not None and mount.credential is not None and not mount.credential.managed
|
owner is not None and owner.credential is not None and not owner.credential.managed
|
||||||
),
|
),
|
||||||
|
copy_from=copy_from,
|
||||||
key_options=_key_options(request),
|
key_options=_key_options(request),
|
||||||
password_options=_password_options(request),
|
password_options=_password_options(request),
|
||||||
)
|
)
|
||||||
@@ -628,16 +633,33 @@ async def mount_edit(request: Request, mount_id: int) -> Response:
|
|||||||
return _mount_form(request, mount_input_from_model(m), {}, m)
|
return _mount_form(request, mount_input_from_model(m), {}, m)
|
||||||
|
|
||||||
|
|
||||||
|
@protected.get("/mounts/{source_id}/copy", response_class=HTMLResponse)
|
||||||
|
async def mount_copy(request: Request, source_id: int) -> Response:
|
||||||
|
"""Formular für eine neue Anbindung, vorausgefüllt aus einer bestehenden."""
|
||||||
|
with nm(request).session_factory() as s:
|
||||||
|
src = s.get(Mount, source_id)
|
||||||
|
if src is None:
|
||||||
|
return redirect("/mounts")
|
||||||
|
data = mount_input_from_model(src)
|
||||||
|
data.name = copy_name(s, src.name)
|
||||||
|
return _mount_form(request, data, {}, copy_from=src)
|
||||||
|
|
||||||
|
|
||||||
@protected.post("/mounts/new", response_class=HTMLResponse)
|
@protected.post("/mounts/new", response_class=HTMLResponse)
|
||||||
@protected.post("/mounts/{mount_id}/edit", response_class=HTMLResponse)
|
@protected.post("/mounts/{mount_id}/edit", response_class=HTMLResponse)
|
||||||
async def mount_save(request: Request, mount_id: int | None = None) -> Response:
|
@protected.post("/mounts/{source_id}/copy", response_class=HTMLResponse)
|
||||||
|
async def mount_save(
|
||||||
|
request: Request, mount_id: int | None = None, source_id: int | None = None
|
||||||
|
) -> Response:
|
||||||
state = nm(request)
|
state = nm(request)
|
||||||
data, errors = parse_mount_form(_form_dict(await request.form()))
|
data, errors = parse_mount_form(_form_dict(await request.form()))
|
||||||
with session_scope(state.session_factory) as s:
|
with session_scope(state.session_factory) as s:
|
||||||
existing = s.get(Mount, mount_id) if mount_id is not None else None
|
copy_from = s.get(Mount, source_id) if source_id is not None else None
|
||||||
if not errors:
|
if not errors:
|
||||||
try:
|
try:
|
||||||
m, errors = save_mount(s, state.secrets, state.mounts, data, mount_id)
|
m, errors = save_mount(
|
||||||
|
s, state.secrets, state.mounts, data, mount_id, copy_from=copy_from
|
||||||
|
)
|
||||||
except AdminError as exc:
|
except AdminError as exc:
|
||||||
errors = {"_": str(exc)}
|
errors = {"_": str(exc)}
|
||||||
m = None
|
m = None
|
||||||
@@ -646,7 +668,10 @@ async def mount_save(request: Request, mount_id: int | None = None) -> Response:
|
|||||||
if errors or m is None:
|
if errors or m is None:
|
||||||
s.rollback()
|
s.rollback()
|
||||||
existing = s.get(Mount, mount_id) if mount_id is not None else None
|
existing = s.get(Mount, mount_id) if mount_id is not None else None
|
||||||
return _mount_form(request, data, errors, existing, status_code=400)
|
copy_from = s.get(Mount, source_id) if source_id is not None else None
|
||||||
|
return _mount_form(
|
||||||
|
request, data, errors, existing, status_code=400, copy_from=copy_from
|
||||||
|
)
|
||||||
saved_id, name = m.id, m.name
|
saved_id, name = m.id, m.name
|
||||||
is_sftp_key = m.type is MountType.SFTP and data.auth is CredentialKind.SSH_KEY
|
is_sftp_key = m.type is MountType.SFTP and data.auth is CredentialKind.SSH_KEY
|
||||||
flash(request, f"Anbindung „{name}“ gespeichert.")
|
flash(request, f"Anbindung „{name}“ gespeichert.")
|
||||||
@@ -668,12 +693,17 @@ async def mount_test_form(request: Request) -> Response:
|
|||||||
data, errors = parse_mount_form(form)
|
data, errors = parse_mount_form(form)
|
||||||
if errors:
|
if errors:
|
||||||
return render(request, "partials/check_result.html", errors=errors, report=None)
|
return render(request, "partials/check_result.html", errors=errors, report=None)
|
||||||
existing_id = int(form["mount_id"]) if form.get("mount_id", "").isdigit() else None
|
ref = form.get("mount_id") or form.get("copy_from") or ""
|
||||||
|
existing_id = int(ref) if ref.isdigit() else None
|
||||||
with state.session_factory() as s:
|
with state.session_factory() as s:
|
||||||
existing = s.get(Mount, existing_id) if existing_id else None
|
existing = s.get(Mount, existing_id) if existing_id else None
|
||||||
cred = credential_for_test(data, existing, state.secrets, s)
|
cred = credential_for_test(data, existing, state.secrets, s)
|
||||||
cfg = data.config()
|
cfg = data.config()
|
||||||
if data.type is MountType.SFTP and existing is not None:
|
if (
|
||||||
|
data.type is MountType.SFTP
|
||||||
|
and existing is not None
|
||||||
|
and (existing.server, existing.port) == (data.server, data.port)
|
||||||
|
):
|
||||||
cfg = dataclasses.replace(cfg, host_key=existing.host_key)
|
cfg = dataclasses.replace(cfg, host_key=existing.host_key)
|
||||||
report = await run_in_threadpool(state.mounts.check, cfg, cred, None)
|
report = await run_in_threadpool(state.mounts.check, cfg, cred, None)
|
||||||
return render(request, "partials/check_result.html", report=report, errors={})
|
return render(request, "partials/check_result.html", report=report, errors={})
|
||||||
@@ -873,6 +903,7 @@ def _rule_form(
|
|||||||
errors: dict[str, str],
|
errors: dict[str, str],
|
||||||
rule_id: int | None,
|
rule_id: int | None,
|
||||||
status_code: int = 200,
|
status_code: int = 200,
|
||||||
|
copy_of: str | None = None,
|
||||||
) -> Response:
|
) -> Response:
|
||||||
with nm(request).session_factory() as s:
|
with nm(request).session_factory() as s:
|
||||||
mounts = s.scalars(select(Mount).order_by(Mount.name)).all()
|
mounts = s.scalars(select(Mount).order_by(Mount.name)).all()
|
||||||
@@ -886,6 +917,7 @@ def _rule_form(
|
|||||||
d=d,
|
d=d,
|
||||||
errors=errors,
|
errors=errors,
|
||||||
rule_id=rule_id,
|
rule_id=rule_id,
|
||||||
|
copy_of=copy_of,
|
||||||
src_mounts=mounts,
|
src_mounts=mounts,
|
||||||
src_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
|
src_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
|
||||||
dst_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
|
dst_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
|
||||||
@@ -913,25 +945,47 @@ async def rule_edit(request: Request, rule_id: int) -> Response:
|
|||||||
return _rule_form(request, d, {}, rule_id)
|
return _rule_form(request, d, {}, rule_id)
|
||||||
|
|
||||||
|
|
||||||
|
@protected.get("/rules/{source_id}/copy", response_class=HTMLResponse)
|
||||||
|
async def rule_copy(request: Request, source_id: int) -> Response:
|
||||||
|
"""Neue Regel, vorausgefüllt aus einer bestehenden – zunächst deaktiviert, damit sie
|
||||||
|
nicht sofort mit dem Original um dieselben Dateien konkurriert."""
|
||||||
|
with nm(request).session_factory() as s:
|
||||||
|
src = s.get(Rule, source_id)
|
||||||
|
if src is None:
|
||||||
|
return redirect("/rules")
|
||||||
|
d = rule_input_from_model(src)
|
||||||
|
d.name = rule_copy_name(s, src.name)
|
||||||
|
d.enabled = False
|
||||||
|
source_name = src.name
|
||||||
|
return _rule_form(request, d, {}, None, copy_of=source_name)
|
||||||
|
|
||||||
|
|
||||||
@protected.post("/rules/new", response_class=HTMLResponse)
|
@protected.post("/rules/new", response_class=HTMLResponse)
|
||||||
@protected.post("/rules/{rule_id}/edit", response_class=HTMLResponse)
|
@protected.post("/rules/{rule_id}/edit", response_class=HTMLResponse)
|
||||||
async def rule_save(request: Request, rule_id: int | None = None) -> Response:
|
@protected.post("/rules/{source_id}/copy", response_class=HTMLResponse)
|
||||||
|
async def rule_save(
|
||||||
|
request: Request, rule_id: int | None = None, source_id: int | None = None
|
||||||
|
) -> Response:
|
||||||
state = nm(request)
|
state = nm(request)
|
||||||
form = _form_dict(await request.form())
|
form = _form_dict(await request.form())
|
||||||
with session_scope(state.session_factory) as s:
|
with session_scope(state.session_factory) as s:
|
||||||
|
source = s.get(Rule, source_id) if source_id is not None else None
|
||||||
|
copy_of = source.name if source is not None else None
|
||||||
d, errors = parse_rule_form(s, form)
|
d, errors = parse_rule_form(s, form)
|
||||||
r = None
|
r = None
|
||||||
if not errors:
|
if not errors:
|
||||||
try:
|
try:
|
||||||
r, errors = save_rule(s, d, rule_id)
|
r, errors = save_rule(s, d, rule_id) # Kopie: rule_id ist None → neue Regel
|
||||||
except AdminError as exc:
|
except AdminError as exc:
|
||||||
errors = {"_": str(exc)}
|
errors = {"_": str(exc)}
|
||||||
if errors or r is None:
|
if errors or r is None:
|
||||||
s.rollback()
|
s.rollback()
|
||||||
return _rule_form(request, d, errors, rule_id, status_code=400)
|
return _rule_form(request, d, errors, rule_id, status_code=400, copy_of=copy_of)
|
||||||
name = r.name
|
name, enabled = r.name, r.enabled
|
||||||
state.scheduler.sync()
|
state.scheduler.sync()
|
||||||
flash(request, f"Regel „{name}“ gespeichert.")
|
flash(request, f"Regel „{name}“ gespeichert.")
|
||||||
|
if copy_of and not enabled:
|
||||||
|
flash(request, "Die Kopie ist deaktiviert. Probelauf machen, dann aktivieren.", "info")
|
||||||
return redirect("/rules")
|
return redirect("/rules")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +1,14 @@
|
|||||||
{% extends "base.html" %}
|
{% extends "base.html" %}
|
||||||
{% from "macros.html" import field, select, check %}
|
{% from "macros.html" import field, select, check %}
|
||||||
{% block title %}{{ 'Anbindung bearbeiten' if mount else 'Neue Anbindung' }}{% endblock %}
|
{% block title %}{{ 'Anbindung bearbeiten' if mount else ('Anbindung kopieren' if copy_from else 'Neue Anbindung') }}{% endblock %}
|
||||||
{% block content %}
|
{% block content %}
|
||||||
<h1>{{ 'Anbindung „' ~ mount.name ~ '“ bearbeiten' if mount else 'Neue Anbindung' }}</h1>
|
<h1>{% if mount %}Anbindung „{{ mount.name }}“ bearbeiten{% elif copy_from %}Kopie von „{{ copy_from.name }}“{% else %}Neue Anbindung{% endif %}</h1>
|
||||||
|
{% if copy_from %}<p class="msg info small">Alle Werte stammen aus „{{ copy_from.name }}“. Zugangsdaten werden übernommen, solange du das Passwort leer lässt – die Kopie bekommt ihr eigenes Exemplar.{% if copy_from.credential and copy_from.credential.public_key and not copy_from.credential.managed %} Der SSH-Schlüssel bleibt derselbe, ein neuer Eintrag in <code>authorized_keys</code> ist nicht nötig.{% endif %}</p>{% endif %}
|
||||||
{% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
|
{% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
|
||||||
<form class="form" method="post" data-dynamic>
|
<form class="form" method="post" data-dynamic>
|
||||||
<input type="hidden" name="csrf_token" value="{{ csrf }}">
|
<input type="hidden" name="csrf_token" value="{{ csrf }}">
|
||||||
{% if mount %}<input type="hidden" name="mount_id" value="{{ mount.id }}">{% endif %}
|
{% if mount %}<input type="hidden" name="mount_id" value="{{ mount.id }}">{% endif %}
|
||||||
|
{% if copy_from %}<input type="hidden" name="copy_from" value="{{ copy_from.id }}">{% endif %}
|
||||||
<fieldset><legend>Verbindung <a class="legend-help" href="/help#anbindung-felder" title="Hilfe">?</a></legend><div class="fields">
|
<fieldset><legend>Verbindung <a class="legend-help" href="/help#anbindung-felder" title="Hilfe">?</a></legend><div class="fields">
|
||||||
{{ field('name', 'Name', d.name, errors=errors, attrs='required maxlength="100"', hint=('Verzeichnis: ' ~ mount.mountpoint if mount and mount.mountpoint else '')) }}
|
{{ field('name', 'Name', d.name, errors=errors, attrs='required maxlength="100"', hint=('Verzeichnis: ' ~ mount.mountpoint if mount and mount.mountpoint else '')) }}
|
||||||
{{ select('type', 'Typ', [('cifs', 'SMB/CIFS'), ('nfs4', 'NFS 4'), ('nfs3', 'NFS 3'), ('sftp', 'SFTP')], d.type.value, errors=errors) }}
|
{{ select('type', 'Typ', [('cifs', 'SMB/CIFS'), ('nfs4', 'NFS 4'), ('nfs3', 'NFS 3'), ('sftp', 'SFTP')], d.type.value, errors=errors) }}
|
||||||
|
|||||||
@@ -22,6 +22,7 @@
|
|||||||
{% endif %}
|
{% endif %}
|
||||||
{% endif %}
|
{% endif %}
|
||||||
<a class="btn small" href="/mounts/{{ m.id }}/edit">Bearbeiten</a>
|
<a class="btn small" href="/mounts/{{ m.id }}/edit">Bearbeiten</a>
|
||||||
|
<a class="btn small" href="/mounts/{{ m.id }}/copy" title="Neue Anbindung mit denselben Werten und Zugangsdaten">Kopieren</a>
|
||||||
<form class="inline" method="post" action="/mounts/{{ m.id }}/delete" data-confirm="„{{ m.name }}“ wirklich löschen?">{{ csrf_field(csrf) }}<button class="small danger">Löschen</button></form>
|
<form class="inline" method="post" action="/mounts/{{ m.id }}/delete" data-confirm="„{{ m.name }}“ wirklich löschen?">{{ csrf_field(csrf) }}<button class="small danger">Löschen</button></form>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
@@ -1,8 +1,11 @@
|
|||||||
{% extends "base.html" %}
|
{% extends "base.html" %}
|
||||||
{% from "macros.html" import field, select, check %}
|
{% from "macros.html" import field, select, check %}
|
||||||
{% block title %}{{ 'Regel bearbeiten' if rule_id else 'Neue Regel' }}{% endblock %}
|
{% block title %}{{ 'Regel bearbeiten' if rule_id else ('Regel kopieren' if copy_of else 'Neue Regel') }}{% endblock %}
|
||||||
{% block content %}
|
{% block content %}
|
||||||
<h1>{{ 'Regel bearbeiten' if rule_id else 'Neue Regel' }}</h1>
|
<h1>{% if rule_id %}Regel bearbeiten{% elif copy_of %}Kopie von „{{ copy_of }}“{% else %}Neue Regel{% endif %}</h1>
|
||||||
|
{% if copy_of %}<p class="msg info small">Alle Einstellungen stammen aus „{{ copy_of }}“. Die Kopie ist zunächst
|
||||||
|
<strong>deaktiviert</strong>, damit sie nicht mit dem Original um dieselben Dateien konkurriert – Quelle bzw. Ziel
|
||||||
|
anpassen, speichern, Probelauf machen, dann aktivieren. Protokoll und Buchführung beginnen bei der Kopie neu.</p>{% endif %}
|
||||||
{% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
|
{% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
|
||||||
{% if not src_mounts %}<p class="msg info">Zuerst eine Anbindung als Quelle <a href="/mounts/new">anlegen</a>.</p>{% endif %}
|
{% if not src_mounts %}<p class="msg info">Zuerst eine Anbindung als Quelle <a href="/mounts/new">anlegen</a>.</p>{% endif %}
|
||||||
<form class="form" method="post" data-dynamic>
|
<form class="form" method="post" data-dynamic>
|
||||||
|
|||||||
@@ -18,6 +18,7 @@
|
|||||||
<button class="small" hx-post="/rules/{{ r.id }}/dryrun" hx-target="#dry-{{ r.id }}" hx-swap="innerHTML">Probelauf <span class="htmx-indicator">…</span></button>
|
<button class="small" hx-post="/rules/{{ r.id }}/dryrun" hx-target="#dry-{{ r.id }}" hx-swap="innerHTML">Probelauf <span class="htmx-indicator">…</span></button>
|
||||||
<form class="inline" method="post" action="/rules/{{ r.id }}/toggle">{{ csrf_field(csrf) }}<button class="small">{{ 'Deaktivieren' if r.enabled else 'Aktivieren' }}</button></form>
|
<form class="inline" method="post" action="/rules/{{ r.id }}/toggle">{{ csrf_field(csrf) }}<button class="small">{{ 'Deaktivieren' if r.enabled else 'Aktivieren' }}</button></form>
|
||||||
<a class="btn small" href="/rules/{{ r.id }}/edit">Bearbeiten</a>
|
<a class="btn small" href="/rules/{{ r.id }}/edit">Bearbeiten</a>
|
||||||
|
<a class="btn small" href="/rules/{{ r.id }}/copy" title="Neue Regel mit denselben Einstellungen">Kopieren</a>
|
||||||
<form class="inline" method="post" action="/rules/{{ r.id }}/delete" data-confirm="Regel „{{ r.name }}“ und ihr Protokoll löschen?">{{ csrf_field(csrf) }}<button class="small danger">Löschen</button></form>
|
<form class="inline" method="post" action="/rules/{{ r.id }}/delete" data-confirm="Regel „{{ r.name }}“ und ihr Protokoll löschen?">{{ csrf_field(csrf) }}<button class="small danger">Löschen</button></form>
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
@@ -0,0 +1,194 @@
|
|||||||
|
"""0.8.0: Anbindungen kopieren – inklusive Zugangsdaten."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
from sqlalchemy import select
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from netmover.admin import MountInput, copy_name, delete_mount, save_mount
|
||||||
|
from netmover.crypto import SecretBox
|
||||||
|
from netmover.models import Credential, CredentialKind, Mount, MountType
|
||||||
|
from netmover.mounting.fake import FakeMountBackend
|
||||||
|
from netmover.mounting.manager import MountManager
|
||||||
|
from tests.webutil import post
|
||||||
|
|
||||||
|
|
||||||
|
def _cifs(**kw: object) -> MountInput:
|
||||||
|
data: dict[str, object] = {
|
||||||
|
"name": "NAS",
|
||||||
|
"type": MountType.CIFS,
|
||||||
|
"server": "10.0.10.6",
|
||||||
|
"share": "Scans",
|
||||||
|
"username": "scanner",
|
||||||
|
"password": "Pa$$wort",
|
||||||
|
}
|
||||||
|
data.update(kw)
|
||||||
|
return MountInput(**data) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_reuses_own_password_as_independent_copy(
|
||||||
|
session: Session, secretbox: SecretBox, manager: MountManager
|
||||||
|
) -> None:
|
||||||
|
src, _ = save_mount(session, secretbox, manager, _cifs())
|
||||||
|
assert src is not None and src.credential is not None
|
||||||
|
copy, err = save_mount(
|
||||||
|
session,
|
||||||
|
secretbox,
|
||||||
|
manager,
|
||||||
|
_cifs(name="Archiv", share="Archiv", password=""),
|
||||||
|
copy_from=src,
|
||||||
|
)
|
||||||
|
assert not err and copy is not None and copy.credential is not None
|
||||||
|
assert copy.credential.id != src.credential.id and not copy.credential.managed
|
||||||
|
assert secretbox.decrypt(copy.credential.secret_enc) == "Pa$$wort"
|
||||||
|
assert copy.credential.username == "scanner"
|
||||||
|
# Vorlage löschen – die Kopie behält ihre Zugangsdaten
|
||||||
|
delete_mount(session, manager, src.id)
|
||||||
|
session.flush()
|
||||||
|
assert secretbox.decrypt(copy.credential.secret_enc) == "Pa$$wort"
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_with_new_password_uses_new_one(
|
||||||
|
session: Session, secretbox: SecretBox, manager: MountManager
|
||||||
|
) -> None:
|
||||||
|
src, _ = save_mount(session, secretbox, manager, _cifs())
|
||||||
|
copy, _ = save_mount(
|
||||||
|
session, secretbox, manager, _cifs(name="B", share="B", password="anders"), copy_from=src
|
||||||
|
)
|
||||||
|
assert copy is not None and copy.credential is not None
|
||||||
|
assert secretbox.decrypt(copy.credential.secret_enc) == "anders"
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_with_stored_credentials_shares_them(
|
||||||
|
session: Session, secretbox: SecretBox, manager: MountManager
|
||||||
|
) -> None:
|
||||||
|
from netmover.credentials import create_password
|
||||||
|
|
||||||
|
cred = create_password(session, secretbox, "NAS-Scanner", "scanner", "x")
|
||||||
|
src, _ = save_mount(
|
||||||
|
session, secretbox, manager, _cifs(username="", password="", cred_id=str(cred.id))
|
||||||
|
)
|
||||||
|
copy, _ = save_mount(
|
||||||
|
session,
|
||||||
|
secretbox,
|
||||||
|
manager,
|
||||||
|
_cifs(name="B", share="B", username="", password="", cred_id=str(cred.id)),
|
||||||
|
copy_from=src,
|
||||||
|
)
|
||||||
|
assert (
|
||||||
|
copy is not None and copy.credential is cred and src is not None and src.credential is cred
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_sftp_keeps_keypair_and_host_key_only_for_same_server(
|
||||||
|
session: Session, secretbox: SecretBox, manager: MountManager
|
||||||
|
) -> None:
|
||||||
|
sftp = {
|
||||||
|
"type": MountType.SFTP,
|
||||||
|
"server": "10.0.10.17",
|
||||||
|
"share": "/consume",
|
||||||
|
"port": 22,
|
||||||
|
"username": "paperless",
|
||||||
|
"auth": CredentialKind.SSH_KEY,
|
||||||
|
"password": "",
|
||||||
|
}
|
||||||
|
src, _ = save_mount(session, secretbox, manager, _cifs(name="Paperless", **sftp))
|
||||||
|
assert src is not None and src.credential is not None
|
||||||
|
src.host_key = "ssh-ed25519 AAAAhost"
|
||||||
|
same, _ = save_mount(
|
||||||
|
session,
|
||||||
|
secretbox,
|
||||||
|
manager,
|
||||||
|
_cifs(name="Paperless 2", **{**sftp, "share": "/consume2"}),
|
||||||
|
copy_from=src,
|
||||||
|
)
|
||||||
|
other, _ = save_mount(
|
||||||
|
session,
|
||||||
|
secretbox,
|
||||||
|
manager,
|
||||||
|
_cifs(name="Anderer", **{**sftp, "server": "10.0.10.99"}),
|
||||||
|
copy_from=src,
|
||||||
|
)
|
||||||
|
assert same is not None and other is not None
|
||||||
|
for m in (same, other):
|
||||||
|
assert m.credential is not None and m.credential.id != src.credential.id
|
||||||
|
assert m.credential.public_key == src.credential.public_key # gleiches Schlüsselpaar
|
||||||
|
assert m.credential.secret_enc == src.credential.secret_enc
|
||||||
|
assert same.host_key == "ssh-ed25519 AAAAhost" and other.host_key is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_name(session: Session, secretbox: SecretBox, manager: MountManager) -> None:
|
||||||
|
save_mount(session, secretbox, manager, _cifs())
|
||||||
|
assert copy_name(session, "NAS") == "NAS (Kopie)"
|
||||||
|
save_mount(session, secretbox, manager, _cifs(name="NAS (Kopie)", share="x"))
|
||||||
|
assert copy_name(session, "NAS") == "NAS (Kopie 2)"
|
||||||
|
|
||||||
|
|
||||||
|
# -- Oberfläche --------------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
MOUNT = {
|
||||||
|
"name": "NAS",
|
||||||
|
"type": "cifs",
|
||||||
|
"server": "nas",
|
||||||
|
"share": "scans",
|
||||||
|
"username": "scanner",
|
||||||
|
"password": "Geheim,123",
|
||||||
|
"auto_mount": "on",
|
||||||
|
"enabled": "on",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _mid(state: Any, name: str) -> int:
|
||||||
|
with state.session_factory() as s:
|
||||||
|
mid = s.scalar(select(Mount.id).where(Mount.name == name))
|
||||||
|
assert mid is not None
|
||||||
|
return mid
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_button_form_test_and_save(
|
||||||
|
client: TestClient, state: Any, backend: FakeMountBackend
|
||||||
|
) -> None:
|
||||||
|
backend.export("nas", "scans")
|
||||||
|
backend.export("nas", "archiv")
|
||||||
|
post(client, "/mounts/new", MOUNT)
|
||||||
|
src_id = _mid(state, "NAS")
|
||||||
|
assert f'href="/mounts/{src_id}/copy"' in client.get("/mounts").text
|
||||||
|
|
||||||
|
form = client.get(f"/mounts/{src_id}/copy").text
|
||||||
|
assert "Kopie von „NAS“" in form and 'value="NAS (Kopie)"' in form
|
||||||
|
assert 'value="scanner"' in form and "Geheim,123" not in form
|
||||||
|
assert f'name="copy_from" value="{src_id}"' in form
|
||||||
|
|
||||||
|
# Testen mit übernommenem Passwort (Feld leer)
|
||||||
|
copy_form = {
|
||||||
|
**MOUNT,
|
||||||
|
"name": "NAS (Kopie)",
|
||||||
|
"share": "archiv",
|
||||||
|
"password": "",
|
||||||
|
"copy_from": str(src_id),
|
||||||
|
}
|
||||||
|
r = post(client, "/mounts/test", copy_form)
|
||||||
|
assert "Einhängen erfolgreich" in r.text
|
||||||
|
spec = [c for c in backend.calls if c.action == "mount"][-1].spec
|
||||||
|
assert spec is not None and spec.credentials is not None
|
||||||
|
assert spec.credentials.password == "Geheim,123"
|
||||||
|
|
||||||
|
r = post(client, f"/mounts/{src_id}/copy", copy_form)
|
||||||
|
assert r.status_code == 303, r.text
|
||||||
|
with state.session_factory() as s:
|
||||||
|
copy = s.get(Mount, _mid(state, "NAS (Kopie)"))
|
||||||
|
assert copy is not None and copy.share == "archiv" and copy.slug == "nas-kopie"
|
||||||
|
assert copy.credential is not None
|
||||||
|
assert state.secrets.decrypt(copy.credential.secret_enc) == "Geheim,123"
|
||||||
|
assert len(s.scalars(select(Credential)).all()) == 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_validation_error_keeps_copy_context(client: TestClient, state: Any) -> None:
|
||||||
|
post(client, "/mounts/new", MOUNT)
|
||||||
|
src_id = _mid(state, "NAS")
|
||||||
|
r = post(client, f"/mounts/{src_id}/copy", {**MOUNT, "name": "NAS", "password": ""})
|
||||||
|
assert r.status_code == 400 and "bereits vergeben" in r.text and "Kopie von „NAS“" in r.text
|
||||||
@@ -0,0 +1,168 @@
|
|||||||
|
"""0.9.0: Regeln kopieren – vorausgefüllt, zunächst deaktiviert."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from netmover.db import session_scope
|
||||||
|
from netmover.models import (
|
||||||
|
ArchiveKind,
|
||||||
|
ArchiveLayout,
|
||||||
|
CopiedFile,
|
||||||
|
Mount,
|
||||||
|
MountType,
|
||||||
|
Rule,
|
||||||
|
RuleAction,
|
||||||
|
Run,
|
||||||
|
)
|
||||||
|
from tests.webutil import post
|
||||||
|
|
||||||
|
|
||||||
|
def _setup(state: Any, tmp_path: Path) -> tuple[int, int]:
|
||||||
|
consume = tmp_path / "consume"
|
||||||
|
consume.mkdir()
|
||||||
|
with session_scope(state.session_factory) as s:
|
||||||
|
m = Mount(
|
||||||
|
name="NAS",
|
||||||
|
slug="nas",
|
||||||
|
type=MountType.CIFS,
|
||||||
|
server="nas",
|
||||||
|
share="scans",
|
||||||
|
mountpoint=str(tmp_path / "mnt" / "nas"),
|
||||||
|
)
|
||||||
|
a = Mount(
|
||||||
|
name="Archiv",
|
||||||
|
slug="archiv",
|
||||||
|
type=MountType.CIFS,
|
||||||
|
server="nas",
|
||||||
|
share="archiv",
|
||||||
|
mountpoint=str(tmp_path / "mnt" / "archiv"),
|
||||||
|
)
|
||||||
|
s.add_all([m, a])
|
||||||
|
s.flush()
|
||||||
|
r = Rule(
|
||||||
|
name="Scans → Paperless",
|
||||||
|
src_mount_id=m.id,
|
||||||
|
src_subpath="max",
|
||||||
|
dst_path=str(consume),
|
||||||
|
pattern="*.pdf;*.jpg",
|
||||||
|
recursive=True,
|
||||||
|
interval_seconds=45,
|
||||||
|
stability_seconds=7,
|
||||||
|
action=RuleAction.COPY,
|
||||||
|
set_mode="0640",
|
||||||
|
quarantine_path="_q",
|
||||||
|
archive_kind=ArchiveKind.MOUNT,
|
||||||
|
archive_mount_id=a.id,
|
||||||
|
archive_path="scans",
|
||||||
|
archive_layout=ArchiveLayout.DATE,
|
||||||
|
)
|
||||||
|
s.add(r)
|
||||||
|
s.flush()
|
||||||
|
s.add(Run(rule_id=r.id))
|
||||||
|
s.add(CopiedFile(rule_id=r.id, src_relpath="a.pdf", size=1, mtime_ns=1, dst_path="x"))
|
||||||
|
return r.id, m.id
|
||||||
|
|
||||||
|
|
||||||
|
def _rule(state: Any, name: str) -> Rule:
|
||||||
|
with state.session_factory() as s:
|
||||||
|
r = s.scalars(select(Rule).where(Rule.name == name)).one()
|
||||||
|
s.expunge(r)
|
||||||
|
return r
|
||||||
|
|
||||||
|
|
||||||
|
def _form_from_page(html: str) -> dict[str, str]:
|
||||||
|
"""Wie ein Browser: alle Felder des Formulars mit ihren vorausgefüllten Werten."""
|
||||||
|
import re
|
||||||
|
|
||||||
|
form: dict[str, str] = {}
|
||||||
|
for m in re.finditer(r'<input[^>]*name="([^"]+)"[^>]*>', html):
|
||||||
|
tag, name = m.group(0), m.group(1)
|
||||||
|
if 'type="checkbox"' in tag:
|
||||||
|
if " checked" in tag:
|
||||||
|
form[name] = "on"
|
||||||
|
else:
|
||||||
|
v = re.search(r'value="([^"]*)"', tag)
|
||||||
|
form[name] = v.group(1) if v else ""
|
||||||
|
for m in re.finditer(r'<select[^>]*name="([^"]+)"[^>]*>(.*?)</select>', html, re.S):
|
||||||
|
sel = re.search(r'<option value="([^"]*)" selected>', m.group(2))
|
||||||
|
form[m.group(1)] = sel.group(1) if sel else ""
|
||||||
|
return form
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_form_is_prefilled_and_disabled(
|
||||||
|
client: TestClient, state: Any, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
rid, _ = _setup(state, tmp_path)
|
||||||
|
assert f'href="/rules/{rid}/copy"' in client.get("/rules").text
|
||||||
|
html = client.get(f"/rules/{rid}/copy").text
|
||||||
|
assert "Kopie von „Scans → Paperless“" in html and "deaktiviert" in html
|
||||||
|
form = _form_from_page(html)
|
||||||
|
assert form["name"] == "Scans → Paperless (Kopie)"
|
||||||
|
assert "enabled" not in form # Häkchen „Aktiv“ nicht gesetzt
|
||||||
|
assert form["pattern"] == "*.pdf;*.jpg" and form["src_subpath"] == "max"
|
||||||
|
assert form["archive_kind"] == "mount" and form["archive_layout"] == "date"
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_saves_new_rule_with_all_settings(
|
||||||
|
client: TestClient, state: Any, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
rid, _ = _setup(state, tmp_path)
|
||||||
|
form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
|
||||||
|
form["src_subpath"] = "erika" # typische Anpassung
|
||||||
|
r = post(client, f"/rules/{rid}/copy", form)
|
||||||
|
assert r.status_code == 303, r.text
|
||||||
|
assert "Die Kopie ist deaktiviert" in client.get("/rules").text
|
||||||
|
|
||||||
|
orig, copy = _rule(state, "Scans → Paperless"), _rule(state, "Scans → Paperless (Kopie)")
|
||||||
|
assert copy.id != orig.id and copy.enabled is False and orig.enabled is True
|
||||||
|
assert copy.src_subpath == "erika" and orig.src_subpath == "max"
|
||||||
|
for f in (
|
||||||
|
"src_mount_id",
|
||||||
|
"dst_path",
|
||||||
|
"pattern",
|
||||||
|
"recursive",
|
||||||
|
"interval_seconds",
|
||||||
|
"stability_seconds",
|
||||||
|
"action",
|
||||||
|
"set_mode",
|
||||||
|
"quarantine_path",
|
||||||
|
"archive_kind",
|
||||||
|
"archive_mount_id",
|
||||||
|
"archive_path",
|
||||||
|
"archive_layout",
|
||||||
|
):
|
||||||
|
assert getattr(copy, f) == getattr(orig, f), f
|
||||||
|
# Protokoll und Buchführung beginnen neu
|
||||||
|
with state.session_factory() as s:
|
||||||
|
assert s.scalars(select(Run).where(Run.rule_id == copy.id)).all() == []
|
||||||
|
assert s.scalars(select(CopiedFile).where(CopiedFile.rule_id == copy.id)).all() == []
|
||||||
|
assert state.scheduler.next_run(copy.id) is None # deaktiviert → kein Termin
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_names_count_up(client: TestClient, state: Any, tmp_path: Path) -> None:
|
||||||
|
rid, _ = _setup(state, tmp_path)
|
||||||
|
for expected in ("Scans → Paperless (Kopie)", "Scans → Paperless (Kopie 2)"):
|
||||||
|
form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
|
||||||
|
assert form["name"] == expected
|
||||||
|
post(client, f"/rules/{rid}/copy", form)
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_validation_error_keeps_context(
|
||||||
|
client: TestClient, state: Any, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
rid, _ = _setup(state, tmp_path)
|
||||||
|
form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
|
||||||
|
form["name"] = "Scans → Paperless" # schon vergeben
|
||||||
|
r = post(client, f"/rules/{rid}/copy", form)
|
||||||
|
assert r.status_code == 400 and "bereits vergeben" in r.text and "Kopie von" in r.text
|
||||||
|
with state.session_factory() as s:
|
||||||
|
assert len(s.scalars(select(Rule)).all()) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_copy_of_missing_rule_redirects(client: TestClient) -> None:
|
||||||
|
assert client.get("/rules/4711/copy").headers["location"] == "/rules"
|
||||||
Reference in New Issue
Block a user