diff --git a/CHANGELOG.md b/CHANGELOG.md index b54335c..5a06ae9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,30 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`. ## [Unreleased] +## [0.11.0] – 2026-10-24 + +### Hinzugefügt +- **Mehrere Ziele je Regel**: neben dem Hauptziel bis zu 9 weitere Ziele (Anbindung + oder lokaler Pfad, je mit Bezeichnung, Ordner und Ablage flach/nach Datum) – z. B. + Rechnung ins DMS, in die Buchhaltung und ins Archiv. Weitere Ziele werden vor dem + Hauptziel beliefert, jedes mit eigener Buchführung; die Quelle wird erst gelöscht, + wenn alle Ziele die Datei haben. +- **Ein nicht erreichbares Ziel hält die anderen nicht auf**: Die erreichbaren Ziele + (auch das Hauptziel) werden beliefert, die Quelle bleibt liegen, bis das fehlende + Ziel zurück ist und nachbeliefert wurde. Die Regel gilt solange als wartend; + Wiederholungen während des Ausfalls erzeugen keine Protokolleinträge. +- Probelauf zeigt für jedes Ziel den künftigen Pfad. +- Im Kopiermodus bekommt ein neu hinzugefügtes Ziel auch die schon vorhandenen Dateien. +- Anleitung: Beispiel „Rechnungseingang: DMS, Buchhaltung und Archiv“. + +### Geändert +- Das Archiv ist jetzt ein weiteres Ziel. Migration 0005 übernimmt vorhandene Archive + als Ziel „Archiv“ samt Buchführung (`rule_targets`, `target_files` statt + `rules.archive_*`, `archived_files`). Downgrade: nur das erste weitere Ziel bleibt. +- Export enthält die Ziele (`targets`); Exporte älterer Versionen mit Archiv werden + beim Import umgewandelt. +- Regelformular: „Ziel“ heißt jetzt „Hauptziel“, „Archiv“ wird zu „Weitere Ziele“. + ## [0.10.0] – 2026-10-23 ### Hinzugefügt @@ -307,7 +331,8 @@ Erste Version. - Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt. -[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.10.0...HEAD +[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.11.0...HEAD +[0.11.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.10.0...v0.11.0 [0.10.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.9.0...v0.10.0 [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 diff --git a/DEVLOG.md b/DEVLOG.md index 7178cde..5530689 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -24,10 +24,38 @@ Zeiten in Ortszeit (Europe/Berlin). | 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) | | 2026-10-23 | 13:42 | 14:10 | 0:28 | – | Sicherung, Kuma-Kennzahlen, CI (0.10.0) | -| **Summe** | | | **4:35** | | | +| 2026-10-24 | 00:10 | 01:00 | 0:50 | – | Mehrere Ziele je Regel (0.11.0) | +| **Summe** | | | **5:25** | | | --- +## 2026-10-24 · 00:10–01:00 · Mehrere Ziele je Regel (0.11.0) + +**Anlass** – Kundenanforderung Rechnungseingang: E-Rechnung ins DMS, in den +Importordner der Buchhaltung **und** ins Archiv – drei Ziele, Ziel + Archiv reichte +nicht mehr. + +**Erledigt** – Tabellen `rule_targets`/`target_files` (Migration 0005, Archiv wird +Ziel „Archiv“), Engine mit Zielliste, Runner prüft Ziele einzeln, `RunResult.waiting_on`, +Formular mit 9 Plätzen (nächster Platz erscheint per `data-show-for`, kein neues JS), +Export/Import inkl. Altformat. 392 Tests. Gegenproben: Quelle trotz fehlendem Ziel +löschen → 3 Tests rot; Buchführung je Ziel weglassen → 6 Tests rot. + +**Entscheidungen** +- *Hauptziel bleibt in `rules`*, nur die weiteren Ziele in einer Tabelle: kleinster + Eingriff ins Schema, bewährte Reihenfolge (weitere Ziele zuerst, Hauptziel zuletzt, + dort weiter der schnelle Umbenennen-Weg) bleibt. +- *Nicht erreichbares Ziel blockiert die anderen nicht*: Bei 3–10 Zielen wäre sonst + jede gestörte Freigabe ein Stillstand. Das Hauptziel bekommt dann eine Kopie + (Buchführung in `copied_files`), die Quelle wird erst nach der Nachlieferung gelöscht. + Gelöscht wird nur in dem Lauf, in dem das letzte Ziel die Datei bekommt. +- *Kein Protokoll-Spam*: Fehlen nur noch nicht erreichbare Ziele, liefert `_process` + None; ein Lauf ohne Einträge wird wie „wartet“ behandelt (kein neuer Run). +- *Beim Speichern Ziele an Ort und Stelle aktualisieren*: Buchführung bleibt, außer + Anbindung oder Ordner eines Ziels ändern sich. +- *Weitere Ziele überschreiben nie* (wie bisher das Archiv) – kein eigenes + Kollisionsfeld je Ziel, weniger Fehlerquellen. + ## 2026-10-23 · 13:42–14:10 · Sicherung, Kuma-Kennzahlen, CI (0.10.0) **Bestandsaufnahme Live-Server** – 0.9.0 seit 15.10. ohne Neustart, 287 Dateien diff --git a/ROADMAP.md b/ROADMAP.md index 99d98aa..c08e3ef 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -17,8 +17,9 @@ Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht ### Weg zu 1.0 -0.10.0 Sicherung + CI → 0.11.0 restliche Punkte oben → ein bis zwei Wochen ruhiger -Betrieb → 1.0.0 ohne neue Funktionen. +0.10.0 Sicherung + CI → 0.11.0 mehrere Ziele (Kundenanforderung; Datenformat muss vor +1.0 feststehen) → 0.12.0 restliche Punkte oben → ein bis zwei Wochen ruhiger Betrieb → +1.0.0 ohne neue Funktionen. **Entschieden:** Kein TLS im internen Netz – HTTP auf Port 8080 ist eine bewusste Entscheidung (README, „Sicherheit“). Der Reverse Proxy bleibt dokumentiert für @@ -31,11 +32,13 @@ nicht vertrauenswürdige Netze. | 💡 | **Eine Regel, mehrere Ziele** je Quell-Unterordner | Heute: eine Regel pro Unterordner | | 💡 | Metriken für Prometheus (`/metrics`) | Ergänzt `/healthz`, z. B. Dateien pro Tag | | 💡 | Deinstallationsskript | Gegenstück zu `install.sh` | +| 💡 | Prüfsumme (SHA-256) je Regel: Kopie zurücklesen und vergleichen | Zusätzlicher Nachweis, z. B. für Rechnungen (GoBD) | ## Erledigt | Version | Thema | |---|---| +| 0.11.0 | Mehrere Ziele je Regel (bis zu 10), unabhängig erreichbar; Archiv wird zum weiteren Ziel | | 0.10.0 | Tägliche verschlüsselte Sicherung + `netmover restore`; Kennzahlen für Kuma; CI in Gitea; Migrationstest-Pflicht | | 0.9.0 | Regeln kopieren (Kopie zunächst deaktiviert) | | 0.8.0 | Anbindungen kopieren inklusive Zugangsdaten | diff --git a/compose.yml b/compose.yml index 42686c1..503280e 100644 --- a/compose.yml +++ b/compose.yml @@ -4,7 +4,7 @@ services: netmover: build: . - image: netmover:0.10.0 + image: netmover:0.11.0 container_name: netmover restart: unless-stopped cap_add: diff --git a/docs/ANLEITUNG.md b/docs/ANLEITUNG.md index 41d7409..e2683c2 100644 --- a/docs/ANLEITUNG.md +++ b/docs/ANLEITUNG.md @@ -8,7 +8,7 @@ Diese Anleitung erklärt die Bedienung. Installation, Updates und technische Det stehen im [README](../README.md). **Inhalt:** [Begriffe](#begriffe) · [Erste Anmeldung](#erste-anmeldung) · -[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) · [Archiv](#archiv) · +[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) · [Weitere Ziele](#weitere-ziele) · [Probelauf](#probelauf) · [Status](#status) · [Wartungsmodus](#wartungsmodus) · [Protokoll](#protokoll) · [Einstellungen](#einstellungen) · [Anmeldung](#anmeldung) · [Uptime Kuma](#uptime-kuma) · [Sicherung](#sicherung) · [Beispiele](#beispiele) · [Häufige Fragen](#fragen) @@ -177,33 +177,48 @@ Gespeicherte Zugangsdaten lassen sich nur löschen, wenn sie keine Anbindung meh | **Aktion** | **verschieben** (Quelle wird nach erfolgreicher Übertragung gelöscht) oder **kopieren** (Quelle bleibt; jede Dateiversion wird nur einmal kopiert). | | **Wenn Datei existiert** | **umbenennen** (`scan.pdf` → `scan-1.pdf`, `scan-2.pdf` …), **überspringen** oder **überschreiben**. | -### Archiv {#archiv} +### Weitere Ziele (Archiv, Buchhaltung, …) {#weitere-ziele} -Optional bekommt jede Datei zusätzlich eine **Kopie ins Archiv** – z. B. Paperless als -Ziel und ein Archivordner auf dem NAS. **Die Quelle wird erst gelöscht, wenn die Datei -im Ziel UND im Archiv angekommen ist.** +Neben dem **Hauptziel** kann jede Regel bis zu **9 weitere +Ziele** beliefern – z. B. eine Rechnung ins DMS (Hauptziel), in den Importordner der +Buchhaltung und ins Archiv. **Die Quelle wird erst gelöscht, wenn alle Ziele die Datei +haben.** | Feld | Bedeutung | |---|---| -| **Archiv** | kein Archiv / eine Anbindung (SMB, NFS, SFTP) / ein lokaler Pfad | -| **Archivordner** | Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren. | -| **Ablage** | **wie in der Quelle** oder **nach Datum** – `2026/10/scan001.pdf` (Monat der Verarbeitung) | +| **Ziel 2 … 10** | Anbindung (SMB, NFS, SFTP) oder lokaler Pfad; „– kein weiteres Ziel –“ = Platz frei. Sobald ein Platz belegt ist, erscheint der nächste. | +| **Bezeichnung** | erscheint in Protokoll und Probelauf, z. B. „Archiv“ oder „Buchhaltung“; leer = Name der Anbindung | +| **Ordner** | Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren. | +| **Ablage** | **wie in der Quelle** oder **nach Datum** – `2026/10/scan001.pdf` (Monat der Verarbeitung). Für Ordner, die eine Software abholt (Buchhaltung), „wie in der Quelle“ wählen. | + +Ein Ziel entfernen: Platz auf „– kein weiteres Ziel –“ stellen und speichern. Die +übrigen rücken nach. So läuft es pro Datei: -1. **Zuerst ins Archiv** – mit denselben Sicherungen wie beim Ziel. -2. **Dann ins Ziel.** +1. **Zuerst die weiteren Ziele** – der Reihe nach, mit denselben Sicherungen wie beim + Hauptziel (`.part`-Datei, Größenprüfung, atomares Umbenennen). +2. **Dann das Hauptziel.** 3. **Dann erst** wird die Quelle gelöscht (bei „verschieben“). -Das Archiv kommt zuerst, weil ein Ziel wie Paperless Dateien sofort abholt: Scheitert -das Archiv, wurde noch nichts geliefert, und die Datei bleibt in der Quelle. Scheitert -danach das Ziel, merkt sich NetMover, dass die Datei schon archiviert ist – der nächste -Versuch legt keine zweite Archivkopie an. +Die weiteren Ziele kommen zuerst, weil ein Hauptziel wie Paperless Dateien sofort +abholt: Scheitert ein weiteres Ziel, ist am Hauptziel noch nichts angekommen, und die +Datei bleibt in der Quelle. NetMover merkt sich je Ziel, welche Datei es schon hat – +ein neuer Versuch legt nirgends eine zweite Kopie an. -Im Archiv wird **nie überschrieben**: Gleiche Namen werden immer umbenannt -(`scan-1.pdf`), egal was die Regel fürs Ziel vorsieht. Ist das Archiv nicht erreichbar, -wartet die Regel wie bei Quelle und Ziel. Im Kopiermodus bekommen Ziel und Archiv je -eine Kopie, die Quelle bleibt. +**Ein Ziel ist nicht erreichbar** (NAS aus, Server weg): Die übrigen Ziele bekommen die +Datei trotzdem, auch das Hauptziel. Die Quelle bleibt liegen, bis das fehlende Ziel +zurück ist; dann wird es nachbeliefert, und erst danach wird die Quelle gelöscht. Die +Regel gilt solange als „wartet“ (Statusseite, `/healthz`, E-Mail beim Ausfall der +Anbindung). Weitere Läufe während des Ausfalls erzeugen keine Protokolleinträge. + +In weiteren Zielen wird **nie überschrieben**: Gleiche Namen werden immer umbenannt +(`scan-1.pdf`), egal was die Regel fürs Hauptziel vorsieht. Im Kopiermodus bekommt jedes +Ziel eine Kopie, die Quelle bleibt; ein neu hinzugefügtes Ziel bekommt dabei auch die +Dateien, die schon in der Quelle liegen. + +Dateien werden byte-genau kopiert – bei E-Rechnungen (ZUGFeRD/Factur-X) bleibt das +eingebettete XML unverändert. ### Sicherheit und Grenzen {#regel-sicherheit} @@ -230,7 +245,7 @@ Synology-Verwaltungsordner (`@eaDir`, `#recycle`). ### 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, +auf **Kopieren** klicken. Alle Einstellungen (Filter, Ziele, Sicherheit, Rechte) sind übernommen, der Name lautet „… (Kopie)“. Die Kopie ist zunächst **deaktiviert**: Mit gleicher Quelle würde sie sonst sofort mit @@ -507,8 +522,8 @@ deaktiviert, Zugangsdaten müssen neu eingegeben werden. 3. Anbindung **Paperless** (SFTP, Pfad `/opt/paperless_data/consume`, Schlüssel wählen, Benutzer = der Paperless-Benutzer), testen. 4. Regel **Scans → Paperless**: Quelle NAS, Filter `*.pdf`, Ziel Paperless, - verschieben, umbenennen, 30 s. Optional **Archiv**: Anbindung NAS, Ordner - `Archiv/Scans`, Ablage nach Datum. **Probelauf**, dann aktivieren. + verschieben, umbenennen, 30 s. Optional als **weiteres Ziel** „Archiv“: Anbindung + NAS, Ordner `Archiv/Scans`, Ablage nach Datum. **Probelauf**, dann aktivieren. ### Jede Person in einen eigenen Ordner {#beispiel-unterordner} @@ -517,6 +532,22 @@ 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. +### Rechnungseingang: DMS, Buchhaltung und Archiv {#beispiel-rechnungen} + +E-Rechnungen (PDF) landen in einem Eingangsordner und sollen ins DMS, in den +Importordner der Buchhaltungssoftware und ins Archiv. + +1. Anbindungen: *Rechnungseingang*, *DMS*, *Buchhaltung*, *Archiv*. +2. Regel: Quelle *Rechnungseingang*, Filter `*.pdf;*.xml` (falls auch reine + XRechnung-XML kommt), **Hauptziel** *DMS*, Aktion verschieben. +3. **Weitere Ziele**: „Buchhaltung“ (Ablage **wie in der Quelle** – die Software sucht + nicht in Unterordnern) und „Archiv“ (Ablage **nach Datum**). +4. **Probelauf** mit ein, zwei Testrechnungen – er zeigt für jedes Ziel den künftigen + Pfad –, dann aktivieren. + +Ist das DMS kurz weg, bekommt die Buchhaltung die Rechnung trotzdem; das DMS wird +nachbeliefert, sobald es zurück ist. + ### Kopie als zusätzliche Sicherung {#beispiel-kopie} Regel mit Aktion **kopieren**, Ziel z. B. ein zweites NAS, Intervall 3600 s. Jede diff --git a/src/netmover/__init__.py b/src/netmover/__init__.py index a7bb35d..706b154 100644 --- a/src/netmover/__init__.py +++ b/src/netmover/__init__.py @@ -1,3 +1,3 @@ """NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben.""" -__version__ = "0.10.0" +__version__ = "0.11.0" diff --git a/src/netmover/admin.py b/src/netmover/admin.py index 4edaab1..3bf84b6 100644 --- a/src/netmover/admin.py +++ b/src/netmover/admin.py @@ -12,14 +12,12 @@ import re from collections.abc import Callable, Mapping from dataclasses import dataclass, field -from sqlalchemy import select +from sqlalchemy import delete, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from netmover.crypto import SecretBox, generate_ssh_keypair from netmover.models import ( - ArchiveKind, - ArchiveLayout, CollisionPolicy, Credential, CredentialKind, @@ -29,6 +27,10 @@ from netmover.models import ( PatternType, Rule, RuleAction, + RuleTarget, + TargetFile, + TargetKind, + TargetLayout, ) from netmover.mounting.manager import CredentialData, MountConfig, MountManager from netmover.validation import ( @@ -50,6 +52,7 @@ from netmover.validation import ( Errors = dict[str, str] MIN_INTERVAL = 10 +MAX_TARGETS = 9 # weitere Ziele zusätzlich zum Hauptziel _OWNER_RE = re.compile(r"^\d{1,10}(:\d{1,10})?$") @@ -462,7 +465,7 @@ def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None: select(Rule.name).where( (Rule.src_mount_id == mount_id) | (Rule.dst_mount_id == mount_id) - | (Rule.archive_mount_id == mount_id) + | Rule.targets.any(RuleTarget.mount_id == mount_id) ) ).all() if used: @@ -480,6 +483,17 @@ def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None: # -- Regeln ---------------------------------------------------------------------------------- +@dataclass +class TargetInput: + """Ein weiteres Ziel im Regelformular (Felder ``target__…``).""" + + name: str = "" + kind: str = "mount" # "mount" | "local" + mount_id: int | None = None + path: str = "" + layout: TargetLayout = TargetLayout.FLAT + + @dataclass class RuleInput: name: str = "" @@ -504,10 +518,7 @@ class RuleInput: dry_run: bool = False quarantine_path: str = "" quarantine_after: int = 3 - archive_kind: ArchiveKind = ArchiveKind.NONE - archive_mount_id: int | None = None - archive_path: str = "" - archive_layout: ArchiveLayout = ArchiveLayout.FLAT + targets: list[TargetInput] = field(default_factory=list) extra: dict[str, str] = field(default_factory=dict) @@ -535,10 +546,9 @@ def rule_input_from_model(r: Rule) -> RuleInput: dry_run=r.dry_run, quarantine_path=r.quarantine_path or "", quarantine_after=r.quarantine_after, - archive_kind=r.archive_kind, - archive_mount_id=r.archive_mount_id, - archive_path=r.archive_path, - archive_layout=r.archive_layout, + targets=[ + TargetInput(t.name, t.kind.value, t.mount_id, t.path, t.layout) for t in r.targets + ], ) @@ -633,41 +643,72 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err except ValidationError as exc: errors["quarantine_path"] = str(exc) - # Archiv - try: - d.archive_kind = ArchiveKind(form.get("archive_kind") or "none") - d.archive_layout = ArchiveLayout(form.get("archive_layout") or "flat") - except ValueError: - errors["archive_kind"] = "Ungültige Auswahl." - d.archive_path = (form.get("archive_path") or "").strip() - if d.archive_kind is ArchiveKind.MOUNT: - d.archive_mount_id = _int(form, "archive_mount_id", errors) - if d.archive_mount_id is None or s.get(Mount, d.archive_mount_id) is None: - errors["archive_mount_id"] = "Archiv-Anbindung wählen." - try: - d.archive_path = safe_relpath("archive_path", d.archive_path) - except ValidationError as exc: - errors["archive_path"] = str(exc) - if ( - "archive_path" not in errors - and d.archive_mount_id == d.dst_mount_id - and d.dst_kind == "mount" - and d.archive_path == d.dst_path - ): - errors["archive_path"] = "Archiv und Ziel sind derselbe Ordner." - elif d.archive_kind is ArchiveKind.LOCAL: - d.archive_mount_id = None - try: - d.archive_path = safe_abspath("archive_path", d.archive_path) - except ValidationError as exc: - errors["archive_path"] = str(exc) - if d.dst_kind == "local" and d.archive_path == d.dst_path: - errors["archive_path"] = "Archiv und Ziel sind derselbe Ordner." - else: - d.archive_mount_id, d.archive_path = None, "" + d.targets = _parse_targets(s, form, d, errors) return d, errors +def _parse_targets( + s: Session, form: Mapping[str, str], d: RuleInput, errors: Errors +) -> list[TargetInput]: + """Weitere Ziele aus den Plätzen ``target_0_…`` bis ``target_8_…``; leere Plätze + (Art „–“) fallen weg, Fehler gelten für die verbleibende Reihenfolge.""" + folders: dict[tuple[int | None, str], str] = {} + if "dst_path" not in errors: + folders[(d.dst_mount_id if d.dst_kind == "mount" else None, d.dst_path)] = "Hauptziel" + result: list[TargetInput] = [] + for slot in range(MAX_TARGETS): + p = f"target_{slot}_" + kind = form.get(p + "kind") or "" + if not kind: + continue + n = len(result) + e = f"target_{n}_" + t = TargetInput( + name=(form.get(p + "name") or "").strip()[:100], + kind=kind, + path=(form.get(p + "path") or "").strip(), + ) + try: + t.layout = TargetLayout(form.get(p + "layout") or "flat") + except ValueError: + errors[e + "layout"] = "Ungültige Auswahl." + if kind == TargetKind.MOUNT: + t.mount_id = _int(form, p + "mount_id", errors) + m = s.get(Mount, t.mount_id) if t.mount_id else None + if m is None: + errors[e + "mount_id"] = "Anbindung wählen." + try: + t.path = safe_relpath(e + "path", t.path) + except ValidationError as exc: + errors[e + "path"] = str(exc) + if ( + t.mount_id == d.src_mount_id + and "src_subpath" not in errors + and t.path == d.src_subpath + ): + errors[e + "path"] = "Quelle und Ziel sind derselbe Ordner." + default_name = m.name if m is not None else "" + elif kind == TargetKind.LOCAL: + try: + t.path = safe_abspath(e + "path", t.path) + except ValidationError as exc: + errors[e + "path"] = str(exc) + default_name = t.path.rstrip("/").rsplit("/", 1)[-1] + else: + errors[e + "kind"] = "Ungültige Zielart." + default_name = "" + t.name = t.name or default_name or f"Ziel {n + 2}" + if e + "path" not in errors: + other = folders.get((t.mount_id, t.path)) + if other: + errors[e + "path"] = f"Derselbe Ordner wie „{other}“." + folders[(t.mount_id, t.path)] = t.name + if any(x.name == t.name for x in result) or t.name == "Hauptziel": + errors[e + "name"] = "Name kommt in dieser Regel schon vor." + result.append(t) + return result + + def rule_copy_name(s: Session, name: str) -> str: """„Scans (Kopie)“, „Scans (Kopie 2)“, … – frei in der Datenbank.""" base = f"{name} (Kopie)"[:100] @@ -678,6 +719,26 @@ def rule_copy_name(s: Session, name: str) -> str: return candidate +def _save_targets(s: Session, r: Rule, targets: list[TargetInput]) -> None: + """Weitere Ziele an Ort und Stelle aktualisieren – so bleibt ihre Buchführung + erhalten. Ändert sich Anbindung oder Ordner eines Ziels, beginnt sie neu.""" + existing = list(r.targets) + for pos, ti in enumerate(targets): + kind = TargetKind(ti.kind) + mount_id = ti.mount_id if kind is TargetKind.MOUNT else None + if pos < len(existing): + t = existing[pos] + if t.id is not None and (t.kind, t.mount_id, t.path) != (kind, mount_id, ti.path): + s.execute(delete(TargetFile).where(TargetFile.target_id == t.id)) + else: + t = RuleTarget() + r.targets.append(t) + t.position, t.name, t.kind, t.mount_id = pos, ti.name, kind, mount_id + t.path, t.layout = ti.path, ti.layout + for t in existing[len(targets) :]: + r.targets.remove(t) + + 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)) if dup is not None and dup != rule_id: @@ -707,12 +768,9 @@ def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rul "on_collision", "dry_run", "quarantine_after", - "archive_kind", - "archive_mount_id", - "archive_path", - "archive_layout", ): setattr(r, f, getattr(d, f)) + _save_targets(s, r, d.targets) r.set_owner = d.set_owner or None r.set_mode = d.set_mode or None r.quarantine_path = d.quarantine_path or None diff --git a/src/netmover/migrations/versions/0005_rule_targets.py b/src/netmover/migrations/versions/0005_rule_targets.py new file mode 100644 index 0000000..697d8b6 --- /dev/null +++ b/src/netmover/migrations/versions/0005_rule_targets.py @@ -0,0 +1,188 @@ +"""rule targets + +Weitere Ziele je Regel (bis zu 9 zusätzlich zum Hauptziel). Ein vorhandenes Archiv +(Migration 0004) wird zum ersten weiteren Ziel „Archiv“, seine Buchführung +(``archived_files``) wandert nach ``target_files``. Downgrade: das erste weitere +Ziel wird wieder zum Archiv, alle weiteren gehen verloren. + +Revision ID: 0005 +Revises: 0004 +Create Date: 2026-10-24 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +import netmover.models + +revision: str = "0005" +down_revision: str | None = "0004" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def _enum(name: str, *values: str) -> sa.Enum: + return sa.Enum(*values, name=name, native_enum=False, create_constraint=True, length=24) + + +def upgrade() -> None: + op.create_table( + "rule_targets", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("rule_id", sa.Integer(), nullable=False), + sa.Column("position", sa.Integer(), nullable=False), + sa.Column("name", sa.String(length=100), nullable=False), + sa.Column("kind", _enum("target_kind", "mount", "local"), nullable=False), + sa.Column("mount_id", sa.Integer(), nullable=True), + sa.Column("path", sa.String(length=1024), nullable=False), + sa.Column("layout", _enum("target_layout", "flat", "date"), nullable=False), + sa.ForeignKeyConstraint( + ["mount_id"], + ["mounts.id"], + name=op.f("fk_rule_targets_mount_id_mounts"), + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["rule_id"], + ["rules.id"], + name=op.f("fk_rule_targets_rule_id_rules"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_rule_targets")), + ) + op.create_index(op.f("ix_rule_targets_rule_id"), "rule_targets", ["rule_id"]) + op.create_table( + "target_files", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("target_id", sa.Integer(), nullable=False), + sa.Column("src_relpath", sa.String(length=2048), nullable=False), + sa.Column("size", sa.BigInteger(), nullable=False), + sa.Column("mtime_ns", sa.BigInteger(), nullable=False), + sa.Column("dst_path", sa.String(length=2048), nullable=False), + sa.Column("delivered_at", netmover.models.UTCDateTime(), nullable=False), + sa.ForeignKeyConstraint( + ["target_id"], + ["rule_targets.id"], + name=op.f("fk_target_files_target_id_rule_targets"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_target_files")), + sa.UniqueConstraint( + "target_id", + "src_relpath", + "size", + "mtime_ns", + name=op.f("uq_target_files_target_id_src_relpath_size_mtime_ns"), + ), + ) + op.create_index(op.f("ix_target_files_target_id"), "target_files", ["target_id"]) + + # Archiv → erstes weiteres Ziel, Buchführung mitnehmen + op.execute( + """ + INSERT INTO rule_targets (rule_id, position, name, kind, mount_id, path, layout) + SELECT id, 0, 'Archiv', archive_kind, + CASE WHEN archive_kind = 'mount' THEN archive_mount_id END, + archive_path, archive_layout + FROM rules WHERE archive_kind != 'none' + """ + ) + op.execute( + """ + INSERT INTO target_files (target_id, src_relpath, size, mtime_ns, dst_path, delivered_at) + SELECT t.id, a.src_relpath, a.size, a.mtime_ns, a.archive_path, a.archived_at + FROM archived_files a JOIN rule_targets t ON t.rule_id = a.rule_id + """ + ) + op.drop_table("archived_files") + with op.batch_alter_table("rules", schema=None) as batch_op: + batch_op.drop_constraint(batch_op.f("fk_rules_archive_mount_id_mounts"), type_="foreignkey") + batch_op.drop_constraint(batch_op.f("ck_rules_archive_kind"), type_="check") + batch_op.drop_constraint(batch_op.f("ck_rules_archive_layout"), type_="check") + batch_op.drop_column("archive_layout") + batch_op.drop_column("archive_path") + batch_op.drop_column("archive_mount_id") + batch_op.drop_column("archive_kind") + + +def downgrade() -> None: + with op.batch_alter_table("rules", schema=None) as batch_op: + batch_op.add_column( + sa.Column( + "archive_kind", + _enum("archive_kind", "none", "mount", "local"), + server_default="none", + nullable=False, + ) + ) + batch_op.add_column(sa.Column("archive_mount_id", sa.Integer(), nullable=True)) + batch_op.add_column( + sa.Column("archive_path", sa.String(length=1024), server_default="", nullable=False) + ) + batch_op.add_column( + sa.Column( + "archive_layout", + _enum("archive_layout", "flat", "date"), + server_default="flat", + nullable=False, + ) + ) + batch_op.create_foreign_key( + batch_op.f("fk_rules_archive_mount_id_mounts"), + "mounts", + ["archive_mount_id"], + ["id"], + ondelete="RESTRICT", + ) + for col, src in ( + ("archive_kind", "kind"), + ("archive_mount_id", "mount_id"), + ("archive_path", "path"), + ("archive_layout", "layout"), + ): + # Spaltennamen sind feste Literale aus dieser Liste + op.execute( + f"UPDATE rules SET {col} = (SELECT t.{src} FROM rule_targets t " # noqa: S608 + "WHERE t.rule_id = rules.id ORDER BY t.position, t.id LIMIT 1) " + "WHERE EXISTS (SELECT 1 FROM rule_targets t WHERE t.rule_id = rules.id)" + ) + op.create_table( + "archived_files", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("rule_id", sa.Integer(), nullable=False), + sa.Column("src_relpath", sa.String(length=2048), nullable=False), + sa.Column("size", sa.BigInteger(), nullable=False), + sa.Column("mtime_ns", sa.BigInteger(), nullable=False), + sa.Column("archive_path", sa.String(length=2048), nullable=False), + sa.Column("archived_at", netmover.models.UTCDateTime(), nullable=False), + sa.ForeignKeyConstraint( + ["rule_id"], + ["rules.id"], + name=op.f("fk_archived_files_rule_id_rules"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_archived_files")), + sa.UniqueConstraint( + "rule_id", + "src_relpath", + "size", + "mtime_ns", + name=op.f("uq_archived_files_rule_id_src_relpath_size_mtime_ns"), + ), + ) + op.execute( + """ + INSERT INTO archived_files (rule_id, src_relpath, size, mtime_ns, archive_path, + archived_at) + SELECT t.rule_id, f.src_relpath, f.size, f.mtime_ns, f.dst_path, f.delivered_at + FROM target_files f JOIN rule_targets t ON t.id = f.target_id + WHERE t.id = (SELECT t2.id FROM rule_targets t2 WHERE t2.rule_id = t.rule_id + ORDER BY t2.position, t2.id LIMIT 1) + """ + ) + op.drop_index(op.f("ix_target_files_target_id"), table_name="target_files") + op.drop_table("target_files") + op.drop_index(op.f("ix_rule_targets_rule_id"), table_name="rule_targets") + op.drop_table("rule_targets") diff --git a/src/netmover/models.py b/src/netmover/models.py index 89c7fd2..c8031b4 100644 --- a/src/netmover/models.py +++ b/src/netmover/models.py @@ -10,7 +10,9 @@ Abweichungen bzw. Ergänzungen gegenüber dem Mindestumfang aus der Spezifikatio * ``credentials.managed`` + ``mounts.username`` (Migration 0002) – Schlüsselverwaltung: gespeicherte SSH-Schlüssel, die mehrere Anbindungen mit jeweils eigenem Benutzernamen nutzen. -* ``rules.archive_*`` + Tabelle ``archived_files`` (Migration 0004) – Archivkopie. +* Tabellen ``rule_targets`` + ``target_files`` (Migration 0005) – weitere Ziele je + Regel (z. B. Archiv, Buchhaltung) mit eigener Buchführung; lösen das Archiv aus + Migration 0004 (``rules.archive_*``, ``archived_files``) ab. * ``rules.waiting_since`` / ``rules.waiting_reason`` (Migration 0003) – Regel wartet auf eine nicht erreichbare Anbindung (automatisch pausiert). * ``mounts.state_since`` – für „hängt seit …“ / „gestört seit …“ auf der Statusseite. @@ -139,13 +141,12 @@ class CollisionPolicy(enum.StrEnum): OVERWRITE = "overwrite" -class ArchiveKind(enum.StrEnum): - NONE = "none" +class TargetKind(enum.StrEnum): MOUNT = "mount" LOCAL = "local" -class ArchiveLayout(enum.StrEnum): +class TargetLayout(enum.StrEnum): FLAT = "flat" # wie in der Quelle DATE = "date" # Jahr/Monat/…, nach Tag der Verarbeitung @@ -271,18 +272,6 @@ class Rule(Base): quarantine_path: Mapped[str | None] = mapped_column(String(1024)) quarantine_after: Mapped[int] = mapped_column(Integer, default=3) - # Archiv (Migration 0004): zusätzliche Kopie; Quelle erst löschen, wenn Ziel UND Archiv da - archive_kind: Mapped[ArchiveKind] = mapped_column( - _enum(ArchiveKind, "archive_kind"), default=ArchiveKind.NONE, server_default="none" - ) - archive_mount_id: Mapped[int | None] = mapped_column( - ForeignKey("mounts.id", ondelete="RESTRICT") - ) - archive_path: Mapped[str] = mapped_column(String(1024), default="", server_default="") - archive_layout: Mapped[ArchiveLayout] = mapped_column( - _enum(ArchiveLayout, "archive_layout"), default=ArchiveLayout.FLAT, server_default="flat" - ) - last_run_at: Mapped[datetime | None] = mapped_column(UTCDateTime) last_success_at: Mapped[datetime | None] = mapped_column(UTCDateTime) consecutive_failures: Mapped[int] = mapped_column(Integer, default=0) @@ -293,7 +282,12 @@ class Rule(Base): src_mount: Mapped[Mount] = relationship(foreign_keys=[src_mount_id]) dst_mount: Mapped[Mount | None] = relationship(foreign_keys=[dst_mount_id]) - archive_mount: Mapped[Mount | None] = relationship(foreign_keys=[archive_mount_id]) + targets: Mapped[list[RuleTarget]] = relationship( + back_populates="rule", + cascade="all, delete-orphan", + passive_deletes=True, + order_by="RuleTarget.position", + ) runs: Mapped[list[Run]] = relationship( back_populates="rule", cascade="all, delete-orphan", passive_deletes=True ) @@ -355,20 +349,44 @@ class CopiedFile(Base): copied_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) -class ArchivedFile(Base): - """Welche Dateiversion schon im Archiv liegt – damit ein Neuversuch nach einem - gescheiterten Ziel keine zweite Archivkopie anlegt.""" +class RuleTarget(Base): + """Weiteres Ziel einer Regel (Archiv, Buchhaltung, …). Wird vor dem Hauptziel + beliefert; die Quelle wird erst gelöscht, wenn alle Ziele die Datei haben.""" - __tablename__ = "archived_files" - __table_args__ = (UniqueConstraint("rule_id", "src_relpath", "size", "mtime_ns"),) + __tablename__ = "rule_targets" id: Mapped[int] = mapped_column(Integer, primary_key=True) - rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE")) + rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE"), index=True) + position: Mapped[int] = mapped_column(Integer, default=0) + name: Mapped[str] = mapped_column(String(100), default="") + kind: Mapped[TargetKind] = mapped_column(_enum(TargetKind, "target_kind")) + # kind=mount: Unterordner relativ zur Anbindung; kind=local: absoluter Pfad + mount_id: Mapped[int | None] = mapped_column(ForeignKey("mounts.id", ondelete="RESTRICT")) + path: Mapped[str] = mapped_column(String(1024), default="") + layout: Mapped[TargetLayout] = mapped_column( + _enum(TargetLayout, "target_layout"), default=TargetLayout.FLAT + ) + + rule: Mapped[Rule] = relationship(back_populates="targets") + mount: Mapped[Mount | None] = relationship() + + +class TargetFile(Base): + """Welche Dateiversion ein weiteres Ziel schon hat – ein Neuversuch (anderes Ziel + gescheitert oder nicht erreichbar) legt so keine zweite Kopie an.""" + + __tablename__ = "target_files" + __table_args__ = (UniqueConstraint("target_id", "src_relpath", "size", "mtime_ns"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + target_id: Mapped[int] = mapped_column( + ForeignKey("rule_targets.id", ondelete="CASCADE"), index=True + ) src_relpath: Mapped[str] = mapped_column(String(2048)) size: Mapped[int] = mapped_column(BigInteger) mtime_ns: Mapped[int] = mapped_column(BigInteger) - archive_path: Mapped[str] = mapped_column(String(2048)) - archived_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) + dst_path: Mapped[str] = mapped_column(String(2048)) + delivered_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) class FileFailure(Base): diff --git a/src/netmover/mover/engine.py b/src/netmover/mover/engine.py index 41b4e58..cebedb8 100644 --- a/src/netmover/mover/engine.py +++ b/src/netmover/mover/engine.py @@ -9,6 +9,12 @@ Reihenfolge pro Lauf: unveränderter Größe **und** mtime gehen weiter (Fallstrick 1), 5. pro Datei: ggf. Quarantäne, Kollisionsregel, Platzprüfung (Größe + 5 %, Fallstrick 14), atomare Übertragung, erst dann Quelle löschen. + +Weitere Ziele (Archiv, Buchhaltung, …) werden **vor** dem Hauptziel beliefert, +jedes mit eigener Buchführung. Die Quelle wird erst gelöscht, wenn alle Ziele die +Datei haben. Ist ein Ziel nicht erreichbar, bekommen die übrigen die Datei trotzdem +(das Hauptziel dann als Kopie); die Quelle bleibt liegen, bis das fehlende Ziel +nachbeliefert ist. """ from __future__ import annotations @@ -17,7 +23,7 @@ import dataclasses import errno import posixpath import time -from collections.abc import Callable +from collections.abc import Callable, Sequence from dataclasses import dataclass, field from datetime import datetime from pathlib import Path @@ -73,7 +79,17 @@ class RuleSnapshot: set_mode: int | None = None quarantine_after: int = 3 dst_root_squash: bool | None = None # nur NFS-Ziele: Ergebnis des Schreibtests - archive_layout: str = "flat" # "flat" | "date" (Jahr/Monat/…) + + +@dataclass(frozen=True) +class Target: + """Weiteres Ziel eines Laufs. ``dest`` None = gerade nicht erreichbar (``reason``).""" + + id: int + name: str + dest: Destination | None + layout: str = "flat" # "flat" | "date" (Jahr/Monat/…) + reason: str = "" @dataclass @@ -97,6 +113,8 @@ class RunResult: items: list[ItemResult] = field(default_factory=list) # Quelle oder Ziel nicht erreichbar → Regel wartet (automatisch pausiert) unavailable: bool = False + # Ziele, die in diesem Lauf nicht erreichbar waren (die übrigen wurden beliefert) + waiting_on: list[str] = field(default_factory=list) def add(self, item: ItemResult) -> None: self.items.append(item) @@ -155,23 +173,27 @@ class Engine: self, rule: RuleSnapshot, source: Source | Path, - dest: Destination, + dest: Destination | None, token: CancelToken, *, dry_run: bool = False, quarantine_rel: str | None = None, quarantine_root: Path | None = None, - archive: Destination | None = None, + targets: Sequence[Target] = (), + dest_reason: str = "", ) -> RunResult: """``source``: eine :class:`Source` oder (Kurzform) ein lokales Verzeichnis. Quarantäne relativ zur Wurzel der Anbindung (``quarantine_rel``) oder – bei - der Kurzform – als Pfad innerhalb des Verzeichnisses (``quarantine_root``).""" + der Kurzform – als Pfad innerhalb des Verzeichnisses (``quarantine_root``). + ``dest`` None = Hauptziel gerade nicht erreichbar (``dest_reason``).""" result = RunResult(dry_run=dry_run) src = source if isinstance(source, Source) else LocalSource(source) if quarantine_root is not None and isinstance(src, LocalSource): quarantine_rel = str(quarantine_root.relative_to(src.root)) try: - self._execute(rule, src, dest, token, result, dry_run, quarantine_rel, archive) + self._execute( + rule, src, dest, token, result, dry_run, quarantine_rel, targets, dest_reason + ) except Cancelled: result.status = RunStatus.ERROR result.message = "Lauf abgebrochen" @@ -193,38 +215,52 @@ class Engine: self, rule: RuleSnapshot, src: Source, - dest: Destination, + dest: Destination | None, token: CancelToken, result: RunResult, dry_run: bool, quarantine_rel: str | None, - archive: Destination | None = None, + targets: Sequence[Target] = (), + dest_reason: str = "", ) -> None: src.check_root() if not dry_run: - dest.prepare() - if archive is not None: - archive.prepare() + if dest is not None: + dest.prepare() + for t in targets: + if t.dest is not None: + t.dest.prepare() token.check() + result.waiting_on = ( + [dest_reason or "Hauptziel nicht erreichbar"] if dest is None else [] + ) + [t.reason or f"„{t.name}“ nicht erreichbar" for t in targets if t.dest is None] exclude = frozenset({quarantine_rel}) if quarantine_rel else frozenset() - if isinstance(dest, LocalDestination) and isinstance(src, LocalSource): - src.exclude_paths.add(dest.root.resolve()) # Ziel innerhalb der Quelle auslassen + if isinstance(src, LocalSource): + for d in (dest, *(t.dest for t in targets)): + if isinstance(d, LocalDestination): + src.exclude_paths.add(d.root.resolve()) # Ziel innerhalb der Quelle auslassen matcher = Matcher(rule.pattern, rule.pattern_type) candidates = src.scan(matcher, rule.recursive, exclude) result.seen = len(candidates) token.check() # Kopier-Buchführung (Fallstrick 11). Gilt auch für „verschieben“, falls - # dort einmal die Quelle nicht gelöscht werden konnte. + # dort einmal die Quelle nicht gelöscht werden konnte oder ein anderes Ziel + # noch fehlt. Dazu je weiteres Ziel, welche Dateiversion es schon hat. done = self.ledger.copied_keys(rule.id) - archived = self.ledger.archived_keys(rule.id) if archive is not None else set() + delivered = {t.id: self.ledger.target_keys(t.id) for t in targets} if not dry_run: present = {c.relpath for c in candidates} self.ledger.prune_copied(rule.id, present) - if archive is not None: - self.ledger.prune_archived(rule.id, present) - fresh = [c for c in candidates if (c.relpath, c.state.size, c.state.mtime_ns) not in done] + for t in targets: + self.ledger.prune_target(t.id, present) + + def complete(c: Candidate) -> bool: + key = (c.relpath, c.state.size, c.state.mtime_ns) + return key in done and all(key in delivered[t.id] for t in targets) + + fresh = [c for c in candidates if not complete(c)] unchanged = len(candidates) - len(fresh) pending: list[Candidate] = [] @@ -281,11 +317,21 @@ class Engine: failures = self.ledger.failures(rule.id) for c in stable: token.check() - result.add( - self._process( - rule, c, src, dest, token, dry_run, quarantine_rel, failures, archive, archived - ) + item = self._process( + rule, + c, + src, + dest, + token, + dry_run, + quarantine_rel, + failures, + targets, + delivered, + done, ) + if item is not None: + result.add(item) if unchanged: result.message = f"{unchanged} bereits übertragen" @@ -318,19 +364,26 @@ class Engine: rule: RuleSnapshot, c: Candidate, src: Source, - dest: Destination, + dest: Destination | None, token: CancelToken, dry_run: bool, quarantine_rel: str | None, failures: dict[str, FailureInfo], - archive: Destination | None = None, - archived: set[tuple[str, int, int]] | None = None, - ) -> ItemResult: + targets: Sequence[Target] = (), + delivered: dict[int, set[tuple[str, int, int]]] | None = None, + done: set[tuple[str, int, int]] | None = None, + ) -> ItemResult | None: + """Eine Datei an alle Ziele, die sie noch nicht haben. None = nichts zu tun, + weil nur noch nicht erreichbare Ziele fehlen (kein Protokolleintrag).""" size = c.state.size - arch_rel = self._archive_rel(c.relpath, rule.archive_layout) - need_archive = archive is not None and ( - (c.relpath, size, c.state.mtime_ns) not in (archived or set()) - ) + key = (c.relpath, size, c.state.mtime_ns) + delivered = delivered or {} + need = [t for t in targets if key not in delivered.get(t.id, set())] + reachable = [(t, t.dest) for t in need if t.dest is not None] + waiting = [t.name for t in need if t.dest is None] + main_needed = key not in (done or set()) + if main_needed and dest is None: + waiting.append("Hauptziel") failure = failures.get(c.relpath) # Quarantäne für wiederholt scheiternde Dateien @@ -350,139 +403,125 @@ class Engine: ) return self._quarantine(rule, c, src, quarantine_rel, failure.attempts) - try: - planned = dest.plan(c.relpath, rule.on_collision) - except OSError as exc: - return self._fail(rule, c, exc, dry_run) - if planned is None: - return ItemResult( - c.relpath, - ItemStatus.SKIPPED, - dest.describe(c.relpath), - size, - "existiert bereits im Ziel", - ) - if dry_run: - msg = ( - "würde überschreiben" - if (rule.on_collision is CollisionPolicy.OVERWRITE and dest.exists(planned)) - else None - ) - verb = "kopieren" if rule.action is RuleAction.COPY else "verschieben" - msg = msg or f"würde {verb}" - if archive is not None: - if need_archive: - arch_planned = archive.plan(arch_rel, CollisionPolicy.RENAME) - msg += f"; Archiv: {archive.describe(arch_planned or arch_rel)}" - else: - msg += "; bereits archiviert" - return ItemResult(c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg) - - try: - free = dest.free_bytes() - except OSError: - free = None - if free is not None and free < size * SPACE_RESERVE: - return ItemResult( - c.relpath, - ItemStatus.FAILED, - dest.describe(planned), - size, - f"zu wenig Platz am Ziel (frei {_mb(free)}, " - f"benötigt {_mb(int(size * SPACE_RESERVE))})", - ) - - if need_archive and archive is not None: + planned: str | None = None + if main_needed and dest is not None: try: - afree = archive.free_bytes() - except OSError: - afree = None - if afree is not None and afree < size * SPACE_RESERVE: + planned = dest.plan(c.relpath, rule.on_collision) + except OSError as exc: + return self._fail(rule, c, exc, dry_run) + if planned is None: return ItemResult( c.relpath, - ItemStatus.FAILED, - archive.describe(arch_rel), + ItemStatus.SKIPPED, + dest.describe(c.relpath), size, - f"zu wenig Platz im Archiv (frei {_mb(afree)}, " - f"benötigt {_mb(int(size * SPACE_RESERVE))})", + "existiert bereits im Ziel", + ) + if dry_run: + return self._plan_item(rule, c, dest, planned, targets, delivered, key) + if not reachable and planned is None: + return None # es fehlen nur Ziele, die gerade nicht erreichbar sind + + if dest is not None and planned is not None: + short = self._no_space(dest, size, "zu wenig Platz am Ziel") + if short: + return ItemResult(c.relpath, ItemStatus.FAILED, dest.describe(planned), size, short) + for t, tdest in reachable: + short = self._no_space(tdest, size, f"{t.name}: zu wenig Platz") + if short: + return ItemResult( + c.relpath, ItemStatus.FAILED, tdest.describe(c.relpath), size, short ) def verify() -> None: if src.state(c.relpath) != c.state: raise SourceChanged("Quelle hat sich während der Übertragung geändert") - # Archiv ZUERST: scheitert es, ist noch nichts ans Ziel (z. B. Paperless) - # geliefert. Scheitert danach das Ziel, verhindert die Buchführung eine - # zweite Archivkopie beim nächsten Versuch. - archive_note: str | None = None - if need_archive and archive is not None: + # Weitere Ziele ZUERST: scheitert eines, ist noch nichts ans Hauptziel (z. B. + # Paperless) geliefert. Scheitert später etwas anderes, verhindert die + # Buchführung eine zweite Kopie beim nächsten Versuch. + notes: list[str] = [] + for t, tdest in reachable: + rel = self._layout_rel(c.relpath, t.layout) try: - meta, _ = self._meta(rule, src, c.relpath, dest) - ares = archive.commit( + meta, _ = self._meta(rule, src, c.relpath, tdest) + tres = tdest.commit( _SourceFile(src, c.relpath), size, - arch_rel, - CollisionPolicy.RENAME, + rel, + CollisionPolicy.RENAME, # weitere Ziele überschreiben nie MetaOptions(meta.atime_ns, meta.mtime_ns, meta.mode, None), verify, token, ) except CollisionSkip: # kann bei RENAME nicht auftreten - return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, "Archiv belegt") + return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, f"{t.name}: belegt") except SourceChanged as exc: return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, str(exc)) except (TransferError, OSError) as exc: - return self._fail(rule, c, exc, dry_run, prefix="Archiv: ") - archive_path = archive.describe(ares.final_rel) - self.ledger.record_archive(rule.id, c.relpath, size, c.state.mtime_ns, archive_path) - archive_note = f"archiviert: {archive_path}" + return self._fail(rule, c, exc, dry_run, prefix=f"{t.name}: ") + path = tdest.describe(tres.final_rel) + self.ledger.record_target(t.id, c.relpath, size, c.state.mtime_ns, path) + notes.append(f"{t.name}: {path}") - try: - meta, notes = self._meta(rule, src, c.relpath, dest) - res = None - local = src.local_path(c.relpath) - if rule.action is RuleAction.MOVE and local is not None: - res = dest.move_same_device(local, c.relpath, rule.on_collision, meta) - if res is None: - res = dest.commit( - _SourceFile(src, c.relpath), - size, + target: str | None = None + consumed = False + if dest is not None and planned is not None: + try: + meta, mnotes = self._meta(rule, src, c.relpath, dest) + res = None + local = src.local_path(c.relpath) + # Schneller Weg (umbenennen) nur, wenn danach kein Ziel mehr fehlt + if rule.action is RuleAction.MOVE and not waiting and local is not None: + res = dest.move_same_device(local, c.relpath, rule.on_collision, meta) + if res is None: + res = dest.commit( + _SourceFile(src, c.relpath), + size, + c.relpath, + rule.on_collision, + meta, + verify, + token, + ) + except CollisionSkip: + return ItemResult( c.relpath, - rule.on_collision, - meta, - verify, - token, + ItemStatus.SKIPPED, + dest.describe(c.relpath), + size, + "existiert bereits im Ziel", ) - except CollisionSkip: - return ItemResult( - c.relpath, - ItemStatus.SKIPPED, - dest.describe(c.relpath), - size, - "existiert bereits im Ziel", - ) - except SourceChanged as exc: - return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, str(exc)) - except (TransferError, OSError) as exc: - return self._fail(rule, c, exc, dry_run) + except SourceChanged as exc: + return ItemResult(c.relpath, ItemStatus.SKIPPED, None, size, str(exc)) + except (TransferError, OSError) as exc: + return self._fail(rule, c, exc, dry_run) + notes += mnotes + res.notes + target = dest.describe(res.final_rel) + consumed = res.source_consumed + if waiting or rule.action is RuleAction.COPY: + self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target) + elif not main_needed: + notes.append("Hauptziel hatte die Datei bereits") - notes += res.notes - if archive_note: - notes.insert(0, archive_note) - target = dest.describe(res.final_rel) if c.relpath in failures: self.ledger.clear_failure(rule.id, c.relpath) + if waiting: + notes.append(f"wartet auf: {', '.join(waiting)} – Quelle bleibt bis dahin") + return ItemResult(c.relpath, ItemStatus.SKIPPED, target, size, "; ".join(notes)) + if rule.action is RuleAction.COPY: - self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target) return ItemResult(c.relpath, ItemStatus.COPIED, target, size, "; ".join(notes) or None) - if not res.source_consumed: + if not consumed: try: src.delete(c.relpath) except OSError as exc: - # Ziel ist vollständig – Buchführung verhindert Duplikate im nächsten Lauf. - self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target) + # Ziele sind vollständig – Buchführung verhindert Duplikate im nächsten Lauf. + self.ledger.record_copy( + rule.id, c.relpath, size, c.state.mtime_ns, target or "(früher geliefert)" + ) return ItemResult( c.relpath, ItemStatus.FAILED, @@ -493,6 +532,49 @@ class Engine: ) return ItemResult(c.relpath, ItemStatus.MOVED, target, size, "; ".join(notes) or None) + def _plan_item( + self, + rule: RuleSnapshot, + c: Candidate, + dest: Destination | None, + planned: str | None, + targets: Sequence[Target], + delivered: dict[int, set[tuple[str, int, int]]], + key: tuple[str, int, int], + ) -> ItemResult: + """Probelauf: was mit der Datei an welchem Ziel passieren würde.""" + verb = "kopieren" if rule.action is RuleAction.COPY else "verschieben" + if dest is None: + parts = ["Hauptziel nicht erreichbar"] + elif planned is None: + parts = ["Hauptziel hat die Datei bereits"] + elif rule.on_collision is CollisionPolicy.OVERWRITE and dest.exists(planned): + parts = ["würde überschreiben"] + else: + parts = [f"würde {verb}"] + for t in targets: + if key in delivered.get(t.id, set()): + parts.append(f"{t.name}: bereits geliefert") + elif t.dest is None: + parts.append(f"{t.name}: nicht erreichbar") + else: + rel = self._layout_rel(c.relpath, t.layout) + parts.append( + f"{t.name}: {t.dest.describe(t.dest.plan(rel, CollisionPolicy.RENAME) or rel)}" + ) + where = dest.describe(planned) if dest is not None and planned is not None else None + return ItemResult(c.relpath, ItemStatus.PLANNED, where, c.state.size, "; ".join(parts)) + + @staticmethod + def _no_space(d: Destination, size: int, label: str) -> str | None: + try: + free = d.free_bytes() + except OSError: + return None + if free is not None and free < size * SPACE_RESERVE: + return f"{label} (frei {_mb(free)}, benötigt {_mb(int(size * SPACE_RESERVE))})" + return None + def _fail( self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool, prefix: str = "" ) -> ItemResult: @@ -507,7 +589,7 @@ class Engine: event(log, "file.failed", level=30, rule=rule.name, file=c.relpath, error=msg) return ItemResult(c.relpath, ItemStatus.FAILED, None, c.state.size, msg) - def _archive_rel(self, relpath: str, layout: str) -> str: + def _layout_rel(self, relpath: str, layout: str) -> str: if layout == "date": t = datetime.fromtimestamp(self._clock()) return f"{t:%Y}/{t:%m}/{relpath}" @@ -545,4 +627,12 @@ class Engine: ) -__all__ = ["Engine", "FileState", "ItemResult", "RuleSnapshot", "RunAborted", "RunResult"] +__all__ = [ + "Engine", + "FileState", + "ItemResult", + "RuleSnapshot", + "RunAborted", + "RunResult", + "Target", +] diff --git a/src/netmover/mover/ledger.py b/src/netmover/mover/ledger.py index 424fdf1..9b7d47a 100644 --- a/src/netmover/mover/ledger.py +++ b/src/netmover/mover/ledger.py @@ -3,6 +3,7 @@ * ``copied_files`` (Fallstrick 11): Welche Dateiversion (Pfad, Größe, mtime) wurde schon übertragen? Verhindert endloses Neu-Kopieren im Kopiermodus – und im Verschiebemodus Duplikate, wenn die Quelle nicht gelöscht werden konnte. +* ``target_files``: welche Dateiversion ein weiteres Ziel schon hat. * ``file_failures``: Fehlversuche pro Datei als Grundlage der Quarantäne. """ @@ -14,7 +15,7 @@ from sqlalchemy import delete, select from sqlalchemy.orm import Session, sessionmaker from netmover.db import session_scope -from netmover.models import ArchivedFile, CopiedFile, FileFailure, utcnow +from netmover.models import CopiedFile, FileFailure, TargetFile, utcnow CopyKey = tuple[str, int, int] # (relpath, size, mtime_ns) @@ -72,51 +73,51 @@ class Ledger: s.execute(delete(CopiedFile).where(CopiedFile.id.in_(stale))) return len(stale) - # -- Archiv ------------------------------------------------------------------------ + # -- weitere Ziele ------------------------------------------------------------------ - def archived_keys(self, rule_id: int) -> set[CopyKey]: + def target_keys(self, target_id: int) -> set[CopyKey]: with session_scope(self.session_factory) as s: rows = s.execute( - select(ArchivedFile.src_relpath, ArchivedFile.size, ArchivedFile.mtime_ns).where( - ArchivedFile.rule_id == rule_id + select(TargetFile.src_relpath, TargetFile.size, TargetFile.mtime_ns).where( + TargetFile.target_id == target_id ) ).all() return {(r[0], r[1], r[2]) for r in rows} - def record_archive( - self, rule_id: int, relpath: str, size: int, mtime_ns: int, archive_path: str + def record_target( + self, target_id: int, relpath: str, size: int, mtime_ns: int, dst: str ) -> None: with session_scope(self.session_factory) as s: exists = s.scalar( - select(ArchivedFile.id).where( - ArchivedFile.rule_id == rule_id, - ArchivedFile.src_relpath == relpath, - ArchivedFile.size == size, - ArchivedFile.mtime_ns == mtime_ns, + select(TargetFile.id).where( + TargetFile.target_id == target_id, + TargetFile.src_relpath == relpath, + TargetFile.size == size, + TargetFile.mtime_ns == mtime_ns, ) ) if exists is None: s.add( - ArchivedFile( - rule_id=rule_id, + TargetFile( + target_id=target_id, src_relpath=relpath, size=size, mtime_ns=mtime_ns, - archive_path=archive_path, + dst_path=dst, ) ) - def prune_archived(self, rule_id: int, present: set[str]) -> int: + def prune_target(self, target_id: int, present: set[str]) -> int: """Wie :meth:`prune_copied`: nur nach einem vollständigen Scan aufrufen.""" with session_scope(self.session_factory) as s: rows = s.execute( - select(ArchivedFile.id, ArchivedFile.src_relpath).where( - ArchivedFile.rule_id == rule_id + select(TargetFile.id, TargetFile.src_relpath).where( + TargetFile.target_id == target_id ) ).all() stale = [r[0] for r in rows if r[1] not in present] if stale: - s.execute(delete(ArchivedFile).where(ArchivedFile.id.in_(stale))) + s.execute(delete(TargetFile).where(TargetFile.id.in_(stale))) return len(stale) def failures(self, rule_id: int) -> dict[str, FailureInfo]: diff --git a/src/netmover/mover/runner.py b/src/netmover/mover/runner.py index 7c4db90..36883ab 100644 --- a/src/netmover/mover/runner.py +++ b/src/netmover/mover/runner.py @@ -23,7 +23,7 @@ from sqlalchemy.orm import Session, sessionmaker from netmover.db import session_scope from netmover.logging_setup import get_logger -from netmover.models import MountType, Rule, RunStatus +from netmover.models import MountType, Rule, RunStatus, TargetKind from netmover.mounting.manager import MountManager, MountUnavailable from netmover.mounting.sftp import SftpError, connect from netmover.mover.destinations import ( @@ -32,7 +32,7 @@ from netmover.mover.destinations import ( SftpDestination, TransferError, ) -from netmover.mover.engine import Engine, RuleSnapshot, RunResult +from netmover.mover.engine import Engine, RuleSnapshot, RunResult, Target from netmover.mover.ledger import Ledger from netmover.mover.sources import LocalSource, SftpSource, Source from netmover.timeouts import CancelToken, OperationTimeout, run_with_watchdog @@ -56,11 +56,18 @@ class RulePlan: quarantine_path: str | None op_timeout: int dry_run: bool - archive_kind: str = "none" - archive_mount_id: int | None = None - archive_type: MountType | None = None - archive_name: str | None = None - archive_path: str = "" + targets: tuple[TargetPlan, ...] = () + + +@dataclass(frozen=True) +class TargetPlan: + id: int + name: str + mount_id: int | None # None = lokaler Pfad + mount_type: MountType | None + mount_name: str | None + path: str + layout: str def parse_owner(value: str | None) -> tuple[int, int] | None: @@ -97,7 +104,6 @@ def load_plan(session: Session, rule_id: int) -> RulePlan: set_mode=parse_mode(r.set_mode), quarantine_after=r.quarantine_after, dst_root_squash=dst.root_squash if dst is not None else None, - archive_layout=r.archive_layout.value, ) return RulePlan( snapshot=snap, @@ -115,14 +121,38 @@ def load_plan(session: Session, rule_id: int) -> RulePlan: else None, op_timeout=r.op_timeout_seconds, dry_run=r.dry_run, - archive_kind=r.archive_kind.value, - archive_mount_id=r.archive_mount_id if r.archive_kind.value == "mount" else None, - archive_type=r.archive_mount.type if r.archive_mount is not None else None, - archive_name=r.archive_mount.name if r.archive_mount is not None else None, - archive_path=r.archive_path, + targets=tuple( + TargetPlan( + id=t.id, + name=t.name or f"Ziel {i + 2}", + mount_id=t.mount_id if t.kind is TargetKind.MOUNT else None, + mount_type=t.mount.type if t.mount is not None else None, + mount_name=t.mount.name if t.mount is not None else None, + path=t.path, + layout=t.layout.value, + ) + for i, t in enumerate(r.targets) + ), ) +def _with_waiting(result: RunResult) -> RunResult: + """Nicht erreichbare Ziele: Ist in diesem Lauf sonst nichts passiert, gilt die + Regel als wartend (wie bei einer fehlenden Quelle, ohne Protokolleintrag); + sonst wird der Lauf normal protokolliert und nennt die fehlenden Ziele.""" + if not result.waiting_on or result.status is RunStatus.ERROR: + return result + reason = "; ".join(result.waiting_on) + if not result.items: + result.status = RunStatus.ERROR + result.unavailable = True + result.message = reason + else: + note = f"wartet auf: {reason}" + result.message = f"{result.message}; {note}" if result.message else note + return result + + class RuleRunner: def __init__( self, @@ -182,44 +212,51 @@ class RuleRunner: ) -> RunResult: try: src_mp = self.mounts.check_alive(plan.src_mount_id) - dst_mp = ( - self.mounts.check_alive(plan.dst_mount_id) - if plan.dst_mount_id is not None - else None - ) - arch_mp = ( - self.mounts.check_alive(plan.archive_mount_id) - if plan.archive_mount_id is not None - else None - ) except MountUnavailable as exc: return RunResult(RunStatus.ERROR, dry, message=str(exc), unavailable=True) + # Ziele einzeln prüfen: ein nicht erreichbares Ziel hält die übrigen nicht auf. + def alive(mount_id: int | None) -> tuple[Path | None, str]: + if mount_id is None: + return None, "" + try: + return self.mounts.check_alive(mount_id), "" + except MountUnavailable as exc: + return None, str(exc) + + dst_mp, dst_down = alive(plan.dst_mount_id) + target_state = [(t, *alive(t.mount_id)) for t in plan.targets] + if dst_down and all(down for _t, _mp, down in target_state): + return RunResult(RunStatus.ERROR, dry, message=dst_down, unavailable=True) + try: make_source = self._source_factory(plan, src_mp) src_root = src_mp / plan.src_subpath if src_mp and plan.src_subpath else src_mp - make_dest = self.destination_factory( - "dst_path", - plan.dst_mount_id, - plan.dst_type, - plan.dst_name, - plan.dst_path, - dst_mp, - src_root, - ) - make_archive = ( - self.destination_factory( - "archive_path", - plan.archive_mount_id, - plan.archive_type, - plan.archive_name, - plan.archive_path, - arch_mp, + make_dest = ( + None + if dst_down + else self.destination_factory( + "dst_path", + plan.dst_mount_id, + plan.dst_type, + plan.dst_name, + plan.dst_path, + dst_mp, src_root, ) - if plan.archive_kind != "none" - else None ) + make_targets = [ + ( + t, + down, + None + if down + else self.destination_factory( + "target_path", t.mount_id, t.mount_type, t.mount_name, t.path, mp, src_root + ), + ) + for t, mp, down in target_state + ] except (ValidationError, TransferError, SftpError) as exc: return RunResult(RunStatus.ERROR, dry, message=str(exc)) @@ -228,23 +265,34 @@ class RuleRunner: try: source = make_source() opened.append(source) - dest = make_dest() - opened.append(dest) - archive = make_archive() if make_archive is not None else None - if archive is not None: - opened.append(archive) - return self.engine.execute( + dest = make_dest() if make_dest is not None else None + if dest is not None: + opened.append(dest) + targets: list[Target] = [] + for t, down, factory in make_targets: + tdest = None + if factory is not None: + try: + tdest = factory() + except (TransferError, SftpError, OSError) as exc: + down = f"„{t.name}“ nicht erreichbar: {exc}" + if tdest is not None: + opened.append(tdest) + targets.append(Target(t.id, t.name, tdest, t.layout, down)) + result = self.engine.execute( plan.snapshot, source, dest, token, dry_run=dry, quarantine_rel=plan.quarantine_path, - archive=archive, + targets=targets, + dest_reason=dst_down, ) finally: for item in reversed(opened): item.close() + return _with_waiting(result) def finish() -> None: self._started.pop(rule_id, None) @@ -302,7 +350,7 @@ class RuleRunner: src_root: Path | None, ) -> Callable[[], Destination]: """Ziel oder Archiv: lokaler Pfad (``mount_id`` None), SFTP oder Kernel-Mount.""" - what = {"archive_path": "Archiv", "backup_path": "Sicherungsziel"}.get(field, "Ziel") + what = {"target_path": "weiteres Ziel", "backup_path": "Sicherungsziel"}.get(field, "Ziel") if mount_id is None: root = Path(safe_abspath(field, path)) if self._under_mount_base(root): diff --git a/src/netmover/runs.py b/src/netmover/runs.py index 8e1d829..9916024 100644 --- a/src/netmover/runs.py +++ b/src/netmover/runs.py @@ -95,6 +95,13 @@ class RunService: message = f"Pausiert, bis die Anbindung zurück ist: {result.message}" update_counters = False # „Fehler in Folge“ nicht hochzählen event(log, "rule.waiting", level=30, rule=rule.name, reason=result.message or "") + elif update_counters and result.waiting_on: + # Ein Ziel fehlt, die übrigen wurden beliefert: Lauf normal + # protokollieren, Regel aber als wartend kennzeichnen (Statusseite, Kuma). + if rule.waiting_since is None: + rule.waiting_since = finished + event(log, "rule.waiting", level=30, rule=rule.name, reason=message or "") + rule.waiting_reason = "; ".join(result.waiting_on) elif ( update_counters and rule.waiting_since is not None diff --git a/src/netmover/transfer.py b/src/netmover/transfer.py index 1eab1e1..36a1416 100644 --- a/src/netmover/transfer.py +++ b/src/netmover/transfer.py @@ -16,8 +16,6 @@ from sqlalchemy.orm import Session from netmover import __version__ from netmover.models import ( - ArchiveKind, - ArchiveLayout, CollisionPolicy, Mount, MountState, @@ -25,6 +23,9 @@ from netmover.models import ( PatternType, Rule, RuleAction, + RuleTarget, + TargetKind, + TargetLayout, ) from netmover.mounting.manager import MountManager from netmover.settings_store import AppSettings, NotifyMode @@ -74,9 +75,6 @@ RULE_FIELDS = ( "dry_run", "quarantine_path", "quarantine_after", - "archive_kind", - "archive_path", - "archive_layout", ) SETTING_FIELDS = ( "retention_days", @@ -104,7 +102,16 @@ def export_config(s: Session, settings: AppSettings) -> dict[str, Any]: **{f: _plain(getattr(r, f)) for f in RULE_FIELDS}, "src_mount": names[r.src_mount_id], "dst_mount": names.get(r.dst_mount_id) if r.dst_mount_id else None, - "archive_mount": names.get(r.archive_mount_id) if r.archive_mount_id else None, + "targets": [ + { + "name": t.name, + "kind": t.kind.value, + "mount": names.get(t.mount_id) if t.mount_id else None, + "path": t.path, + "layout": t.layout.value, + } + for t in r.targets + ], } for r in s.scalars(select(Rule).order_by(Rule.name)) ], @@ -171,16 +178,14 @@ def import_config(s: Session, mounts: MountManager, data: dict[str, Any]) -> Imp try: src = ids[raw["src_mount"]] dst = ids[raw["dst_mount"]] if raw.get("dst_mount") else None - arch = ids[raw["archive_mount"]] if raw.get("archive_mount") else None - r = Rule(src_mount_id=src, dst_mount_id=dst, archive_mount_id=arch) + r = Rule(src_mount_id=src, dst_mount_id=dst) for f in RULE_FIELDS: if f in raw: setattr(r, f, raw[f]) r.pattern_type = PatternType(r.pattern_type) r.action = RuleAction(r.action) r.on_collision = CollisionPolicy(r.on_collision) - r.archive_kind = ArchiveKind(r.archive_kind or "none") - r.archive_layout = ArchiveLayout(r.archive_layout or "flat") + r.targets = _import_targets(raw, ids) r.src_subpath = safe_relpath("src_subpath", r.src_subpath or "") if int(r.interval_seconds) < 10: raise ValueError("Intervall < 10 s") @@ -198,6 +203,35 @@ def import_config(s: Session, mounts: MountManager, data: dict[str, Any]) -> Imp return report +def _import_targets(raw: dict[str, Any], ids: dict[str, int]) -> list[RuleTarget]: + """Weitere Ziele; Exporte vor 0.11.0 kennen nur ein Archiv (``archive_*``).""" + items = raw.get("targets") + if items is None and raw.get("archive_kind") not in (None, "none"): + items = [ + { + "name": "Archiv", + "kind": raw["archive_kind"], + "mount": raw.get("archive_mount"), + "path": raw.get("archive_path") or "", + "layout": raw.get("archive_layout") or "flat", + } + ] + targets = [] + for pos, t in enumerate(items or []): + kind = TargetKind(t["kind"]) + targets.append( + RuleTarget( + position=pos, + name=str(t.get("name") or f"Ziel {pos + 2}")[:100], + kind=kind, + mount_id=ids[t["mount"]] if kind is TargetKind.MOUNT else None, + path=str(t.get("path") or ""), + layout=TargetLayout(t.get("layout") or "flat"), + ) + ) + return targets + + def apply_imported_settings(current: AppSettings, values: dict[str, Any]) -> AppSettings: """Nach dem Commit des Imports aufrufen (eigene Transaktion).""" for key, value in values.items(): diff --git a/src/netmover/web/app.py b/src/netmover/web/app.py index 2a61bd6..4611734 100644 --- a/src/netmover/web/app.py +++ b/src/netmover/web/app.py @@ -35,6 +35,7 @@ from starlette.middleware.sessions import SessionMiddleware from netmover import __version__ from netmover.admin import ( + MAX_TARGETS, AdminError, MountInput, RuleInput, @@ -79,6 +80,7 @@ from netmover.models import ( PatternType, Rule, RuleAction, + RuleTarget, Run, RunItem, RunStatus, @@ -161,6 +163,7 @@ def _duration(start: datetime | None, end: datetime | None) -> str: TEMPLATES = Jinja2Templates(directory=str(files("netmover.web") / "templates")) TEMPLATES.env.filters.update(ago=_ago, localdt=_localdt, bytes=_bytes) +TEMPLATES.env.globals["max_targets"] = MAX_TARGETS TEMPLATES.env.globals.update( version=__version__, duration=_duration, @@ -452,14 +455,9 @@ class RuleView: target: str -def _archive_target(r: Rule) -> str | None: - if r.archive_kind.value == "none": - return None - if r.archive_kind.value == "local" or r.archive_mount is None: - place = r.archive_path - else: - place = f"{r.archive_mount.name}:/{r.archive_path}" - return place + (" (nach Datum)" if r.archive_layout.value == "date" else "") +def _extra_target(t: RuleTarget) -> str: + place = t.path if t.kind.value == "local" or t.mount is None else f"{t.mount.name}:/{t.path}" + return f"{t.name} {place}" + (" (nach Datum)" if t.layout.value == "date" else "") def _rule_target(r: Rule) -> str: @@ -468,8 +466,7 @@ def _rule_target(r: Rule) -> str: dst = r.dst_path else: dst = f"{r.dst_mount.name}:/{r.dst_path}" if r.dst_path else f"{r.dst_mount.name}:/" - archive = _archive_target(r) - return f"{src} → {dst}" + (f" + Archiv {archive}" if archive else "") + return f"{src} → {dst}" + "".join(f" + {_extra_target(t)}" for t in r.targets) async def _status_context(request: Request) -> dict[str, Any]: diff --git a/src/netmover/web/static/app.css b/src/netmover/web/static/app.css index c846de4..13f4583 100644 --- a/src/netmover/web/static/app.css +++ b/src/netmover/web/static/app.css @@ -176,3 +176,6 @@ details.rename[open] > form { position: absolute; right: 0; top: calc(100% + 4px padding: 8px; background: var(--card); border: 1px solid var(--line); border-radius: var(--radius); box-shadow: 0 6px 18px rgba(0,0,0,.12); flex-wrap: nowrap; } input.w-auto { width: auto; flex: 1; min-width: 200px; max-width: 360px; } +/* Weitere Ziele im Regelformular: je Platz eine eigene Zeile */ +.target-slot { padding-top: 10px; margin-top: 10px; border-top: 1px dashed var(--line); } +.target-slot:first-of-type { border-top: 0; margin-top: 4px; padding-top: 0; } diff --git a/src/netmover/web/templates/rule_form.html b/src/netmover/web/templates/rule_form.html index 06130cd..b2a58dd 100644 --- a/src/netmover/web/templates/rule_form.html +++ b/src/netmover/web/templates/rule_form.html @@ -25,7 +25,7 @@ {{ check('recursive', 'Unterordner einbeziehen (Struktur bleibt erhalten)', d.recursive) }} -
Ziel ?
+
Hauptziel ?
{{ select('dst_kind', 'Zielart', [('mount', 'Anbindung'), ('local', 'Lokaler Pfad')], d.dst_kind, errors=errors) }}
{{ select('dst_mount_id', 'Anbindung', dst_options, d.dst_mount_id or '', errors=errors) }}
{{ field('dst_path', 'Zielordner', d.dst_path, errors=errors, hint='Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren.') }} @@ -33,12 +33,22 @@ {{ select('on_collision', 'Wenn Datei existiert', [('rename', 'umbenennen (scan-1.pdf)'), ('skip', 'überspringen'), ('overwrite', 'überschreiben')], d.on_collision.value, errors=errors) }}
-
Archiv ?
- {{ select('archive_kind', 'Archiv', [('none', 'kein Archiv'), ('mount', 'Anbindung'), ('local', 'Lokaler Pfad')], d.archive_kind.value, errors=errors, hint='Zusätzliche Kopie. Die Quelle wird erst gelöscht, wenn Ziel UND Archiv sie haben.') }} -
{{ select('archive_mount_id', 'Archiv-Anbindung', dst_options, d.archive_mount_id or '', errors=errors) }}
-
{{ field('archive_path', 'Archivordner', d.archive_path, errors=errors, hint='Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren.') }}
-
{{ select('archive_layout', 'Ablage', [('flat', 'wie in der Quelle'), ('date', 'nach Datum (Jahr/Monat)')], d.archive_layout.value, errors=errors, hint='Im Archiv wird nie überschrieben – gleiche Namen werden umbenannt.') }}
-
+
Weitere Ziele ? +

Zusätzliche Kopien, z. B. Archiv oder Buchhaltung – bis zu {{ max_targets }}. Sie werden vor dem Hauptziel beliefert; + die Quelle wird erst gelöscht, wenn alle Ziele die Datei haben. Weitere Ziele überschreiben nie, gleiche Namen werden umbenannt.

+ {% for i in range(max_targets) %} + {% set t = d.targets[i] if i < d.targets|length else none %} +
0 %} data-show-for="target_{{ i - 1 }}_kind:mount local|target_{{ i }}_kind:mount local"{% endif %}> +
+ {{ select('target_' ~ i ~ '_kind', 'Ziel ' ~ (i + 2), [('', '– kein weiteres Ziel –'), ('mount', 'Anbindung'), ('local', 'Lokaler Pfad')], t.kind if t else '', errors=errors) }} +
{{ field('target_' ~ i ~ '_name', 'Bezeichnung', t.name if t else '', errors=errors, placeholder='z. B. Archiv, Buchhaltung', attrs='maxlength="100"', hint='erscheint im Protokoll; leer = Name der Anbindung') }}
+
{{ select('target_' ~ i ~ '_mount_id', 'Anbindung', dst_options, (t.mount_id if t and t.mount_id else ''), errors=errors) }}
+
{{ field('target_' ~ i ~ '_path', 'Ordner', t.path if t else '', errors=errors, hint='Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren.') }}
+
{{ select('target_' ~ i ~ '_layout', 'Ablage', [('flat', 'wie in der Quelle'), ('date', 'nach Datum (Jahr/Monat)')], t.layout.value if t else 'flat', errors=errors) }}
+
+
+ {% endfor %} +
Sicherheit und Grenzen ?
{{ field('stability_seconds', 'Stabilitätsfenster (s)', d.stability_seconds, type='number', errors=errors, attrs='min="0"', hint='Größe und Zeit müssen so lange gleich bleiben') }} diff --git a/tests/test_archive.py b/tests/test_archive.py deleted file mode 100644 index 2f3b289..0000000 --- a/tests/test_archive.py +++ /dev/null @@ -1,351 +0,0 @@ -"""0.7.0: Archiv – Kopie ans Ziel UND ins Archiv, Quelle erst danach löschen.""" - -from __future__ import annotations - -import os -from collections.abc import Iterator -from datetime import datetime -from pathlib import Path - -import pytest -from sqlalchemy import select -from sqlalchemy.orm import Session, sessionmaker - -from netmover.db import session_scope -from netmover.models import ( - ArchivedFile, - ArchiveKind, - CollisionPolicy, - ItemStatus, - Rule, - RuleAction, - RunStatus, -) -from netmover.mounting.fake import FakeMountBackend -from netmover.mounting.manager import MountManager -from netmover.mover.destinations import LocalDestination, TransferError -from netmover.mover.engine import Engine, RuleSnapshot, RunResult -from netmover.mover.ledger import Ledger -from netmover.mover.runner import RuleRunner -from netmover.timeouts import CancelToken -from tests.conftest import MountFactory - - -@pytest.fixture -def rule_id(make_mount: MountFactory, session_factory: sessionmaker[Session]) -> int: - mid = make_mount() - with session_scope(session_factory) as s: - r = Rule(name="r", src_mount_id=mid, dst_path="/x") - s.add(r) - s.flush() - return r.id - - -FIXED = datetime(2026, 10, 12, 9, 30).timestamp() - - -@pytest.fixture -def engine(session_factory: sessionmaker[Session]) -> Engine: - return Engine(Ledger(session_factory), sleep=lambda _: None, clock=lambda: FIXED) - - -@pytest.fixture -def dirs(tmp_path: Path) -> tuple[Path, Path, Path]: - src, dst, arch = tmp_path / "src", tmp_path / "dst", tmp_path / "archiv" - for d in (src, dst, arch): - d.mkdir() - return src, dst, arch - - -@pytest.fixture(autouse=True) -def cross_device(monkeypatch: pytest.MonkeyPatch) -> None: - """tmp_path liegt auf einem Gerät – realistischer Weg über Kopie erzwingen.""" - monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) - - -def snap(rule_id: int, **kw: object) -> RuleSnapshot: - return RuleSnapshot(id=rule_id, name="r", stability_seconds=0, **kw) # type: ignore[arg-type] - - -def run( - engine: Engine, rule: RuleSnapshot, src: Path, dst: Path, arch: Path | None, **kw: object -) -> RunResult: - archive = LocalDestination(arch) if arch is not None else None - return engine.execute(rule, src, LocalDestination(dst), CancelToken(), archive=archive, **kw) # type: ignore[arg-type] - - -def write(p: Path, data: bytes = b"inhalt") -> Path: - p.parent.mkdir(parents=True, exist_ok=True) - p.write_bytes(data) - return p - - -def names(d: Path) -> list[str]: - return sorted(str(p.relative_to(d)) for p in d.rglob("*") if p.is_file()) - - -def test_move_delivers_to_target_and_archive_then_deletes_source( - engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] -) -> None: - src, dst, arch = dirs - data = os.urandom(200_000) - write(src / "scan001.pdf", data) - result = run(engine, snap(rule_id), src, dst, arch) - assert result.status is RunStatus.OK - item = result.items[0] - assert item.status is ItemStatus.MOVED and item.message and "archiviert:" in item.message - assert (dst / "scan001.pdf").read_bytes() == data == (arch / "scan001.pdf").read_bytes() - assert not (src / "scan001.pdf").exists() - assert names(dst) == ["scan001.pdf"] and names(arch) == ["scan001.pdf"] # keine .part-Reste - - -def test_date_layout(engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]) -> None: - src, dst, arch = dirs - write(src / "max" / "a.pdf") - run(engine, snap(rule_id, recursive=True, archive_layout="date"), src, dst, arch) - assert names(arch) == ["2026/10/max/a.pdf"] and names(dst) == ["max/a.pdf"] - - -def test_archive_failure_delivers_nothing_and_keeps_source( - engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch -) -> None: - src, dst, arch = dirs - write(src / "a.pdf") - real = LocalDestination._link_noreplace - - def fail_in_archive(self: LocalDestination, tmp: str, final: str) -> None: - if self.root == arch: - raise TransferError("Archiv voll") - real(self, tmp, final) - - monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_archive) - result = run(engine, snap(rule_id), src, dst, arch) - item = result.items[0] - assert item.status is ItemStatus.FAILED and item.message == "Archiv: Archiv voll (Versuch 1)" - assert (src / "a.pdf").exists() and names(dst) == [] and names(arch) == [] - - -def test_target_failure_after_archive_never_archives_twice( - engine: Engine, - rule_id: int, - dirs: tuple[Path, Path, Path], - monkeypatch: pytest.MonkeyPatch, - session_factory: sessionmaker[Session], -) -> None: - src, dst, arch = dirs - write(src / "a.pdf") - real = LocalDestination._link_noreplace - - def fail_in_target(self: LocalDestination, tmp: str, final: str) -> None: - if self.root == dst: - raise TransferError("Ziel gesperrt") - real(self, tmp, final) - - monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_target) - first = run(engine, snap(rule_id), src, dst, arch) - assert first.items[0].status is ItemStatus.FAILED - assert (src / "a.pdf").exists() and names(arch) == ["a.pdf"] and names(dst) == [] - with session_factory() as s: - assert len(s.scalars(select(ArchivedFile)).all()) == 1 - - monkeypatch.setattr(LocalDestination, "_link_noreplace", real) # Ziel wieder frei - second = run(engine, snap(rule_id), src, dst, arch) - assert second.items[0].status is ItemStatus.MOVED - assert names(arch) == ["a.pdf"] # keine zweite Archivkopie (a-1.pdf) - assert names(dst) == ["a.pdf"] and not (src / "a.pdf").exists() - - -def test_archive_never_overwrites( - engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] -) -> None: - src, dst, arch = dirs - write(arch / "scan.pdf", b"alt im Archiv") - write(src / "scan.pdf", b"neu") - run(engine, snap(rule_id, on_collision=CollisionPolicy.OVERWRITE), src, dst, arch) - assert (arch / "scan.pdf").read_bytes() == b"alt im Archiv" - assert (arch / "scan-1.pdf").read_bytes() == b"neu" - - -def test_no_space_in_archive_writes_nothing( - engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch -) -> None: - src, dst, arch = dirs - write(src / "a.pdf", b"x" * 1000) - real = LocalDestination.free_bytes - monkeypatch.setattr( - LocalDestination, "free_bytes", lambda self: 10 if self.root == arch else real(self) - ) - result = run(engine, snap(rule_id), src, dst, arch) - assert "zu wenig Platz im Archiv" in (result.items[0].message or "") - assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists() - - -def test_source_changed_during_archive_copy( - engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch -) -> None: - src, dst, arch = dirs - f = write(src / "a.pdf", b"original") - real = LocalDestination._apply_meta - - def change(self: LocalDestination, rel: str, meta: object) -> list[str]: - if self.root == arch: - f.write_bytes(b"geaendert!!") - return real(self, rel, meta) # type: ignore[arg-type] - - monkeypatch.setattr(LocalDestination, "_apply_meta", change) - result = run(engine, snap(rule_id), src, dst, arch) - assert result.items[0].status is ItemStatus.SKIPPED - assert names(dst) == [] and names(arch) == [] and f.exists() - - -def test_copy_mode_with_archive( - engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] -) -> None: - src, dst, arch = dirs - f = write(src / "a.pdf", b"v1") - rule = snap(rule_id, action=RuleAction.COPY) - for _ in range(3): - run(engine, rule, src, dst, arch) - assert f.exists() and names(dst) == ["a.pdf"] and names(arch) == ["a.pdf"] - f.write_bytes(b"version 2") - run(engine, rule, src, dst, arch) - assert names(dst) == ["a-1.pdf", "a.pdf"] and names(arch) == ["a-1.pdf", "a.pdf"] - - -def test_dry_run_shows_archive_and_writes_nothing( - engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] -) -> None: - src, dst, arch = dirs - write(src / "a.pdf") - result = run(engine, snap(rule_id, archive_layout="date"), src, dst, arch, dry_run=True) - assert result.items[0].status is ItemStatus.PLANNED - assert f"Archiv: {arch / '2026/10/a.pdf'}" in (result.items[0].message or "") - assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists() - - -def test_without_archive_unchanged( - engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] -) -> None: - src, dst, arch = dirs - write(src / "a.pdf") - result = run(engine, snap(rule_id), src, dst, None) - assert result.items[0].message is None and names(arch) == [] - - -# -- Regellauf ---------------------------------------------------------------------------------- - - -@pytest.fixture -def setup( - manager: MountManager, - fake_backend: FakeMountBackend, - make_mount: MountFactory, - session_factory: sessionmaker[Session], - tmp_path: Path, -) -> Iterator[tuple[int, Path, Path, Path]]: - scans = fake_backend.export("nas", "scans") - arch_remote = fake_backend.export("nas", "archiv") - src_id = make_mount("scans", share="scans") - arch_id = make_mount("archiv", share="archiv") - manager.mount(src_id) - manager.mount(arch_id) - consume = tmp_path / "consume" - consume.mkdir() - with session_scope(session_factory) as s: - r = Rule( - name="Scans → Paperless + Archiv", - src_mount_id=src_id, - dst_path=str(consume), - stability_seconds=0, - archive_kind=ArchiveKind.MOUNT, - archive_mount_id=arch_id, - archive_path="scans", - ) - s.add(r) - s.flush() - rid = r.id - yield rid, scans, consume, arch_remote - - -def test_runner_with_archive_mount( - setup: tuple[int, Path, Path, Path], - manager: MountManager, - session_factory: sessionmaker[Session], -) -> None: - rid, scans, consume, arch = setup - write(scans / "doc.pdf", b"%PDF") - result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid) - assert result.status is RunStatus.OK, result.message - assert ( - (consume / "doc.pdf").read_bytes() == b"%PDF" == (arch / "scans" / "doc.pdf").read_bytes() - ) - assert not (scans / "doc.pdf").exists() - - -def test_unavailable_archive_makes_rule_wait( - setup: tuple[int, Path, Path, Path], - manager: MountManager, - fake_backend: FakeMountBackend, - session_factory: sessionmaker[Session], -) -> None: - rid, scans, consume, _arch = setup - write(scans / "doc.pdf") - fake_backend.vanish(manager.mountpoint_for("archiv")) - result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid) - assert result.unavailable and result.status is RunStatus.ERROR - assert (scans / "doc.pdf").exists() and list(consume.iterdir()) == [] - - -def test_mount_used_as_archive_cannot_be_deleted( - setup: tuple[int, Path, Path, Path], - session_factory: sessionmaker[Session], - manager: MountManager, -) -> None: - from netmover.admin import AdminError, delete_mount - from netmover.models import Mount - - with session_scope(session_factory) as s: - arch_id = s.scalar(select(Mount.id).where(Mount.name == "archiv")) - assert arch_id is not None - with pytest.raises(AdminError, match="Wird noch von Regeln benutzt"): - delete_mount(s, manager, arch_id) - - -def test_rule_form_archive_validation( - session_factory: sessionmaker[Session], make_mount: MountFactory -) -> None: - from netmover.admin import parse_rule_form, save_rule - - mid = make_mount() - base = { - "name": "x", - "src_mount_id": str(mid), - "dst_kind": "local", - "dst_path": "/srv/consume", - "interval_seconds": "30", - "stability_seconds": "5", - "op_timeout_seconds": "120", - } - with session_scope(session_factory) as s: - _, err = parse_rule_form( - s, {**base, "archive_kind": "local", "archive_path": "/srv/consume"} - ) - assert "derselbe Ordner" in err["archive_path"] - _, err = parse_rule_form(s, {**base, "archive_kind": "mount", "archive_path": "a"}) - assert "archive_mount_id" in err - _, err = parse_rule_form(s, {**base, "archive_kind": "local", "archive_path": "relativ"}) - assert "absoluten Pfad" in err["archive_path"] - d, err = parse_rule_form( - s, - { - **base, - "archive_kind": "mount", - "archive_mount_id": str(mid), - "archive_path": "archiv/scans", - "archive_layout": "date", - }, - ) - assert err == {} - r, _ = save_rule(s, d) - assert r is not None and r.archive_kind is ArchiveKind.MOUNT - assert r.archive_path == "archiv/scans" and r.archive_layout.value == "date" diff --git a/tests/test_copy_rule.py b/tests/test_copy_rule.py index ce69d0f..03ab981 100644 --- a/tests/test_copy_rule.py +++ b/tests/test_copy_rule.py @@ -10,14 +10,15 @@ from sqlalchemy import select from netmover.db import session_scope from netmover.models import ( - ArchiveKind, - ArchiveLayout, CopiedFile, Mount, MountType, Rule, RuleAction, + RuleTarget, Run, + TargetKind, + TargetLayout, ) from tests.webutil import post @@ -56,10 +57,17 @@ def _setup(state: Any, tmp_path: Path) -> tuple[int, int]: 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, + targets=[ + RuleTarget( + position=0, + name="Archiv", + kind=TargetKind.MOUNT, + mount_id=a.id, + path="scans", + layout=TargetLayout.DATE, + ), + RuleTarget(position=1, name="Buchhaltung", kind=TargetKind.LOCAL, path="/srv/buha"), + ], ) s.add(r) s.flush() @@ -71,7 +79,8 @@ def _setup(state: Any, tmp_path: Path) -> tuple[int, int]: 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) + _ = list(r.targets) # vor dem Lösen aus der Session laden + s.expunge_all() return r @@ -105,7 +114,9 @@ def test_copy_form_is_prefilled_and_disabled( 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" + assert form["target_0_kind"] == "mount" and form["target_0_layout"] == "date" + assert form["target_0_name"] == "Archiv" and form["target_1_path"] == "/srv/buha" + assert form["target_2_kind"] == "" def test_copy_saves_new_rule_with_all_settings( @@ -131,12 +142,14 @@ def test_copy_saves_new_rule_with_all_settings( "action", "set_mode", "quarantine_path", - "archive_kind", - "archive_mount_id", - "archive_path", - "archive_layout", ): assert getattr(copy, f) == getattr(orig, f), f + + def targets(r: Rule) -> list[tuple[object, ...]]: + return [(t.position, t.name, t.kind, t.mount_id, t.path, t.layout) for t in r.targets] + + assert len(targets(copy)) == 2 and targets(copy) == targets(orig) + assert {t.id for t in copy.targets}.isdisjoint(t.id for t in orig.targets) # Protokoll und Buchführung beginnen neu with state.session_factory() as s: assert s.scalars(select(Run).where(Run.rule_id == copy.id)).all() == [] diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 1d17e34..893f1b9 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -219,3 +219,56 @@ def test_migration_files_are_named_after_their_revision() -> None: versions = Path(__file__).parent.parent / "src/netmover/migrations/versions" names = sorted(p.name.split("_", 1)[0] for p in versions.glob("[0-9]*.py")) assert names == _revisions() + + +def test_0005_turns_archive_into_first_target_and_back(tmp_path: Path) -> None: + """Archiv (0004) samt Buchführung wird zum weiteren Ziel „Archiv“ – und zurück.""" + db = tmp_path / "v09.db" + url = f"sqlite:///{db}" + upgrade(url, "0001") + _seed(db) + upgrade(url, "0004") + c = sqlite3.connect(db) + c.executescript( + """ + UPDATE rules SET archive_kind = 'mount', archive_mount_id = 1, + archive_path = 'archiv', archive_layout = 'date'; + INSERT INTO archived_files (rule_id, src_relpath, size, mtime_ns, archive_path, + archived_at) + VALUES (1, 'a.pdf', 3, 7, '/mnt/netmover/nas/archiv/2026/10/a.pdf', + '2026-10-12 09:00:00'); + """ + ) + c.commit() + c.close() + + upgrade(url, "0005") + c = sqlite3.connect(db) + assert c.execute( + "SELECT rule_id, position, name, kind, mount_id, path, layout FROM rule_targets" + ).fetchall() == [(1, 0, "Archiv", "mount", 1, "archiv", "date")] + assert c.execute("SELECT src_relpath, size, mtime_ns FROM target_files").fetchall() == [ + ("a.pdf", 3, 7) + ] + cols = [r[1] for r in c.execute("PRAGMA table_info(rules)")] + assert not [col for col in cols if col.startswith("archive_")] + assert ( + c.execute("SELECT name FROM sqlite_master WHERE name = 'archived_files'").fetchall() == [] + ) + # zweites Ziel – geht beim Downgrade verloren, das erste bleibt + c.execute( + "INSERT INTO rule_targets (rule_id, position, name, kind, mount_id, path, layout) " + "VALUES (1, 1, 'Buchhaltung', 'local', NULL, '/srv/buha', 'flat')" + ) + c.commit() + c.close() + assert _state(db)["fk_violations"] == [] + + command.downgrade(alembic_config(url), "0004") + c = sqlite3.connect(db) + assert c.execute( + "SELECT archive_kind, archive_mount_id, archive_path, archive_layout FROM rules" + ).fetchall() == [("mount", 1, "archiv", "date")] + assert c.execute("SELECT rule_id, src_relpath FROM archived_files").fetchall() == [(1, "a.pdf")] + c.close() + assert _state(db)["fk_violations"] == [] diff --git a/tests/test_targets.py b/tests/test_targets.py new file mode 100644 index 0000000..4371ac4 --- /dev/null +++ b/tests/test_targets.py @@ -0,0 +1,683 @@ +"""Weitere Ziele (0.11.0, vorher Archiv aus 0.7.0): Kopie an alle Ziele, Quelle erst +danach löschen; nicht erreichbare Ziele werden nachbeliefert.""" + +from __future__ import annotations + +import os +from collections.abc import Iterator +from datetime import datetime +from pathlib import Path +from typing import Any + +import pytest +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from netmover.db import session_scope +from netmover.models import ( + CollisionPolicy, + ItemStatus, + Rule, + RuleAction, + RuleTarget, + RunStatus, + TargetFile, + TargetKind, +) +from netmover.mounting.fake import FakeMountBackend +from netmover.mounting.manager import MountManager +from netmover.mover.destinations import LocalDestination, TransferError +from netmover.mover.engine import Engine, RuleSnapshot, RunResult, Target +from netmover.mover.ledger import Ledger +from netmover.mover.runner import RuleRunner +from netmover.timeouts import CancelToken +from tests.conftest import MountFactory + + +@pytest.fixture +def rule_id(make_mount: MountFactory, session_factory: sessionmaker[Session]) -> int: + mid = make_mount() + with session_scope(session_factory) as s: + r = Rule( + name="r", + src_mount_id=mid, + dst_path="/x", + targets=[ + RuleTarget(position=i, name=n, kind=TargetKind.LOCAL, path=f"/{n}") + for i, n in enumerate(("Archiv", "Buchhaltung", "Extra")) + ], + ) + s.add(r) + s.flush() + return r.id + + +@pytest.fixture +def tids(rule_id: int, session_factory: sessionmaker[Session]) -> dict[str, int]: + with session_factory() as s: + return {t.name: t.id for t in s.scalars(select(RuleTarget))} + + +FIXED = datetime(2026, 10, 12, 9, 30).timestamp() + + +@pytest.fixture +def engine(session_factory: sessionmaker[Session]) -> Engine: + return Engine(Ledger(session_factory), sleep=lambda _: None, clock=lambda: FIXED) + + +@pytest.fixture +def dirs(tmp_path: Path) -> tuple[Path, Path, Path]: + src, dst, arch = tmp_path / "src", tmp_path / "dst", tmp_path / "archiv" + for d in (src, dst, arch): + d.mkdir() + return src, dst, arch + + +@pytest.fixture(autouse=True) +def cross_device(monkeypatch: pytest.MonkeyPatch) -> None: + """tmp_path liegt auf einem Gerät – realistischer Weg über Kopie erzwingen.""" + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + + +def snap(rule_id: int, **kw: object) -> RuleSnapshot: + return RuleSnapshot(id=rule_id, name="r", stability_seconds=0, **kw) # type: ignore[arg-type] + + +LAYOUT: dict[str, str] = {} + + +def run( + engine: Engine, + rule: RuleSnapshot, + src: Path, + dst: Path | None, + arch: Path | None, + *, + layout: str = "flat", + extra: list[Target] | None = None, + **kw: object, +) -> RunResult: + """Hauptziel ``dst`` (None = nicht erreichbar) und Archiv ``arch`` (Ziel „Archiv“).""" + targets = list(extra or []) + if arch is not None: + targets.insert(0, Target(_TIDS["Archiv"], "Archiv", LocalDestination(arch), layout)) + dest = LocalDestination(dst) if dst is not None else None + return engine.execute(rule, src, dest, CancelToken(), targets=targets, **kw) # type: ignore[arg-type] + + +_TIDS: dict[str, int] = {} + + +@pytest.fixture(autouse=True) +def _register_tids(request: pytest.FixtureRequest) -> None: + if "rule_id" in request.fixturenames: + _TIDS.clear() + _TIDS.update(request.getfixturevalue("tids")) + + +def write(p: Path, data: bytes = b"inhalt") -> Path: + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(data) + return p + + +def names(d: Path) -> list[str]: + return sorted(str(p.relative_to(d)) for p in d.rglob("*") if p.is_file()) + + +def test_move_delivers_to_target_and_archive_then_deletes_source( + engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] +) -> None: + src, dst, arch = dirs + data = os.urandom(200_000) + write(src / "scan001.pdf", data) + result = run(engine, snap(rule_id), src, dst, arch) + assert result.status is RunStatus.OK + item = result.items[0] + assert item.status is ItemStatus.MOVED and item.message and "Archiv: " in item.message + assert (dst / "scan001.pdf").read_bytes() == data == (arch / "scan001.pdf").read_bytes() + assert not (src / "scan001.pdf").exists() + assert names(dst) == ["scan001.pdf"] and names(arch) == ["scan001.pdf"] # keine .part-Reste + + +def test_date_layout(engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path]) -> None: + src, dst, arch = dirs + write(src / "max" / "a.pdf") + run(engine, snap(rule_id, recursive=True), src, dst, arch, layout="date") + assert names(arch) == ["2026/10/max/a.pdf"] and names(dst) == ["max/a.pdf"] + + +def test_archive_failure_delivers_nothing_and_keeps_source( + engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + src, dst, arch = dirs + write(src / "a.pdf") + real = LocalDestination._link_noreplace + + def fail_in_archive(self: LocalDestination, tmp: str, final: str) -> None: + if self.root == arch: + raise TransferError("Archiv voll") + real(self, tmp, final) + + monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_archive) + result = run(engine, snap(rule_id), src, dst, arch) + item = result.items[0] + assert item.status is ItemStatus.FAILED and item.message == "Archiv: Archiv voll (Versuch 1)" + assert (src / "a.pdf").exists() and names(dst) == [] and names(arch) == [] + + +def test_target_failure_after_archive_never_archives_twice( + engine: Engine, + rule_id: int, + dirs: tuple[Path, Path, Path], + monkeypatch: pytest.MonkeyPatch, + session_factory: sessionmaker[Session], +) -> None: + src, dst, arch = dirs + write(src / "a.pdf") + real = LocalDestination._link_noreplace + + def fail_in_target(self: LocalDestination, tmp: str, final: str) -> None: + if self.root == dst: + raise TransferError("Ziel gesperrt") + real(self, tmp, final) + + monkeypatch.setattr(LocalDestination, "_link_noreplace", fail_in_target) + first = run(engine, snap(rule_id), src, dst, arch) + assert first.items[0].status is ItemStatus.FAILED + assert (src / "a.pdf").exists() and names(arch) == ["a.pdf"] and names(dst) == [] + with session_factory() as s: + assert len(s.scalars(select(TargetFile)).all()) == 1 + + monkeypatch.setattr(LocalDestination, "_link_noreplace", real) # Ziel wieder frei + second = run(engine, snap(rule_id), src, dst, arch) + assert second.items[0].status is ItemStatus.MOVED + assert names(arch) == ["a.pdf"] # keine zweite Archivkopie (a-1.pdf) + assert names(dst) == ["a.pdf"] and not (src / "a.pdf").exists() + + +def test_archive_never_overwrites( + engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] +) -> None: + src, dst, arch = dirs + write(arch / "scan.pdf", b"alt im Archiv") + write(src / "scan.pdf", b"neu") + run(engine, snap(rule_id, on_collision=CollisionPolicy.OVERWRITE), src, dst, arch) + assert (arch / "scan.pdf").read_bytes() == b"alt im Archiv" + assert (arch / "scan-1.pdf").read_bytes() == b"neu" + + +def test_no_space_in_archive_writes_nothing( + engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + src, dst, arch = dirs + write(src / "a.pdf", b"x" * 1000) + real = LocalDestination.free_bytes + monkeypatch.setattr( + LocalDestination, "free_bytes", lambda self: 10 if self.root == arch else real(self) + ) + result = run(engine, snap(rule_id), src, dst, arch) + assert "Archiv: zu wenig Platz" in (result.items[0].message or "") + assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists() + + +def test_source_changed_during_archive_copy( + engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + src, dst, arch = dirs + f = write(src / "a.pdf", b"original") + real = LocalDestination._apply_meta + + def change(self: LocalDestination, rel: str, meta: object) -> list[str]: + if self.root == arch: + f.write_bytes(b"geaendert!!") + return real(self, rel, meta) # type: ignore[arg-type] + + monkeypatch.setattr(LocalDestination, "_apply_meta", change) + result = run(engine, snap(rule_id), src, dst, arch) + assert result.items[0].status is ItemStatus.SKIPPED + assert names(dst) == [] and names(arch) == [] and f.exists() + + +def test_copy_mode_with_archive( + engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] +) -> None: + src, dst, arch = dirs + f = write(src / "a.pdf", b"v1") + rule = snap(rule_id, action=RuleAction.COPY) + for _ in range(3): + run(engine, rule, src, dst, arch) + assert f.exists() and names(dst) == ["a.pdf"] and names(arch) == ["a.pdf"] + f.write_bytes(b"version 2") + run(engine, rule, src, dst, arch) + assert names(dst) == ["a-1.pdf", "a.pdf"] and names(arch) == ["a-1.pdf", "a.pdf"] + + +def test_dry_run_shows_archive_and_writes_nothing( + engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] +) -> None: + src, dst, arch = dirs + write(src / "a.pdf") + result = run(engine, snap(rule_id), src, dst, arch, layout="date", dry_run=True) + assert result.items[0].status is ItemStatus.PLANNED + assert f"Archiv: {arch / '2026/10/a.pdf'}" in (result.items[0].message or "") + assert names(dst) == [] and names(arch) == [] and (src / "a.pdf").exists() + + +def test_without_archive_unchanged( + engine: Engine, rule_id: int, dirs: tuple[Path, Path, Path] +) -> None: + src, dst, arch = dirs + write(src / "a.pdf") + result = run(engine, snap(rule_id), src, dst, None) + assert result.items[0].message is None and names(arch) == [] + + +# -- mehrere Ziele ------------------------------------------------------------------------------ + + +def _three(tmp_path: Path, rule_id: int, tids: dict[str, int], down: str = "") -> list[Target]: + """Archiv, Buchhaltung, Extra als lokale Ordner; ``down`` = nicht erreichbar.""" + out = [] + for name in ("Archiv", "Buchhaltung", "Extra"): + d = tmp_path / name.lower() + d.mkdir(exist_ok=True) + dest = None if name == down else LocalDestination(d) + out.append(Target(tids[name], name, dest, reason=f"„{name}“ nicht erreichbar")) + return out + + +def test_four_targets_all_get_the_file_then_source_is_deleted( + engine: Engine, rule_id: int, tids: dict[str, int], tmp_path: Path +) -> None: + src, dst = tmp_path / "src", tmp_path / "dst" + src.mkdir() + dst.mkdir() + data = os.urandom(50_000) + write(src / "rechnung.pdf", data) + targets = _three(tmp_path, rule_id, tids) + result = run(engine, snap(rule_id), src, dst, None, extra=targets) + assert result.status is RunStatus.OK and result.waiting_on == [] + item = result.items[0] + assert item.status is ItemStatus.MOVED + for name in ("Archiv", "Buchhaltung", "Extra"): + assert f"{name}: " in (item.message or "") + assert (tmp_path / name.lower() / "rechnung.pdf").read_bytes() == data + assert (dst / "rechnung.pdf").read_bytes() == data + assert not (src / "rechnung.pdf").exists() + + +def test_unreachable_target_others_are_served_and_caught_up_later( + engine: Engine, + rule_id: int, + tids: dict[str, int], + tmp_path: Path, + session_factory: sessionmaker[Session], +) -> None: + src, dst = tmp_path / "src", tmp_path / "dst" + src.mkdir() + dst.mkdir() + write(src / "r1.pdf", b"%PDF-1") + down = _three(tmp_path, rule_id, tids, down="Buchhaltung") + + first = run(engine, snap(rule_id), src, dst, None, extra=down) + assert first.waiting_on == ["„Buchhaltung“ nicht erreichbar"] + item = first.items[0] + assert item.status is ItemStatus.SKIPPED and "wartet auf: Buchhaltung" in (item.message or "") + # Hauptziel und erreichbare Ziele haben die Datei, die Quelle bleibt + assert (dst / "r1.pdf").exists() and (tmp_path / "archiv" / "r1.pdf").exists() + assert (tmp_path / "extra" / "r1.pdf").exists() and (src / "r1.pdf").exists() + assert not (tmp_path / "buchhaltung" / "r1.pdf").exists() + + # Weitere Läufe während des Ausfalls: nichts doppelt, kein Protokolleintrag + again = run(engine, snap(rule_id), src, dst, None, extra=down) + assert again.items == [] and names(dst) == ["r1.pdf"] + assert names(tmp_path / "archiv") == ["r1.pdf"] + + # Buchhaltung wieder da → nachbeliefert, erst jetzt Quelle gelöscht + back = run(engine, snap(rule_id), src, dst, None, extra=_three(tmp_path, rule_id, tids)) + item = back.items[0] + assert item.status is ItemStatus.MOVED, item.message + assert "Hauptziel hatte die Datei bereits" in (item.message or "") + assert names(tmp_path / "buchhaltung") == ["r1.pdf"] + assert names(dst) == ["r1.pdf"] and names(tmp_path / "archiv") == ["r1.pdf"] + assert not (src / "r1.pdf").exists() + + +def test_unreachable_main_target_other_targets_still_served( + engine: Engine, rule_id: int, tids: dict[str, int], tmp_path: Path +) -> None: + src, dst = tmp_path / "src", tmp_path / "dst" + src.mkdir() + dst.mkdir() + write(src / "r.pdf") + targets = _three(tmp_path, rule_id, tids) + first = run(engine, snap(rule_id), src, None, None, extra=targets, dest_reason="DMS weg") + assert first.waiting_on == ["DMS weg"] + assert "wartet auf: Hauptziel" in (first.items[0].message or "") + assert names(tmp_path / "archiv") == ["r.pdf"] and (src / "r.pdf").exists() + second = run(engine, snap(rule_id), src, dst, None, extra=targets) + assert second.items[0].status is ItemStatus.MOVED + assert names(dst) == ["r.pdf"] and names(tmp_path / "extra") == ["r.pdf"] + assert not (src / "r.pdf").exists() + + +def test_copy_mode_new_target_gets_backfilled( + engine: Engine, rule_id: int, tids: dict[str, int], tmp_path: Path +) -> None: + src, dst = tmp_path / "src", tmp_path / "dst" + src.mkdir() + dst.mkdir() + write(src / "a.pdf") + rule = snap(rule_id, action=RuleAction.COPY) + targets = _three(tmp_path, rule_id, tids) + run(engine, rule, src, dst, None, extra=targets[:1]) + run(engine, rule, src, dst, None, extra=targets[:1]) + assert names(tmp_path / "archiv") == ["a.pdf"] and names(dst) == ["a.pdf"] + # Ziel hinzugefügt: nur das neue Ziel bekommt die Datei + result = run(engine, rule, src, dst, None, extra=targets[:2]) + assert result.items[0].status is ItemStatus.COPIED + assert names(tmp_path / "buchhaltung") == ["a.pdf"] + assert names(dst) == ["a.pdf"] and names(tmp_path / "archiv") == ["a.pdf"] + + +def test_dry_run_lists_every_target( + engine: Engine, rule_id: int, tids: dict[str, int], tmp_path: Path +) -> None: + src, dst = tmp_path / "src", tmp_path / "dst" + src.mkdir() + dst.mkdir() + write(src / "a.pdf") + targets = _three(tmp_path, rule_id, tids, down="Extra") + result = run(engine, snap(rule_id), src, dst, None, extra=targets, dry_run=True) + msg = result.items[0].message or "" + assert ( + "würde verschieben" in msg and "Buchhaltung: " in msg and "Extra: nicht erreichbar" in msg + ) + assert (src / "a.pdf").exists() and names(dst) == [] + + +# -- Regellauf ---------------------------------------------------------------------------------- + + +@pytest.fixture +def setup( + manager: MountManager, + fake_backend: FakeMountBackend, + make_mount: MountFactory, + session_factory: sessionmaker[Session], + tmp_path: Path, +) -> Iterator[tuple[int, Path, Path, Path, Path]]: + scans = fake_backend.export("nas", "scans") + arch_remote = fake_backend.export("nas", "archiv") + buha_remote = fake_backend.export("nas", "buha") + src_id = make_mount("scans", share="scans") + arch_id = make_mount("archiv", share="archiv") + buha_id = make_mount("buha", share="buha") + for mid in (src_id, arch_id, buha_id): + manager.mount(mid) + consume = tmp_path / "consume" + consume.mkdir() + with session_scope(session_factory) as s: + r = Rule( + name="Rechnungen → DMS + Buchhaltung + Archiv", + src_mount_id=src_id, + dst_path=str(consume), + stability_seconds=0, + targets=[ + RuleTarget(position=0, name="Buchhaltung", kind=TargetKind.MOUNT, mount_id=buha_id), + RuleTarget( + position=1, + name="Archiv", + kind=TargetKind.MOUNT, + mount_id=arch_id, + path="rechnungen", + ), + ], + ) + s.add(r) + s.flush() + rid = r.id + yield rid, scans, consume, arch_remote, buha_remote + + +def test_runner_with_three_targets( + setup: tuple[int, Path, Path, Path, Path], + manager: MountManager, + session_factory: sessionmaker[Session], +) -> None: + rid, scans, consume, arch, buha = setup + write(scans / "re.pdf", b"%PDF") + result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid) + assert result.status is RunStatus.OK, result.message + for path in (consume / "re.pdf", arch / "rechnungen" / "re.pdf", buha / "re.pdf"): + assert path.read_bytes() == b"%PDF" + assert not (scans / "re.pdf").exists() + + +def test_runner_unavailable_target_waits_without_blocking_others( + setup: tuple[int, Path, Path, Path, Path], + manager: MountManager, + fake_backend: FakeMountBackend, + session_factory: sessionmaker[Session], +) -> None: + from netmover.runs import RunService + + rid, scans, consume, arch, buha = setup + write(scans / "re.pdf") + fake_backend.vanish(manager.mountpoint_for("buha")) + fake_backend.fail_next_mount(times=50) + runner = RuleRunner(session_factory, manager, sleep=lambda _: None) + service = RunService(session_factory, runner) + service.execute(rid) + assert (consume / "re.pdf").exists() and (arch / "rechnungen" / "re.pdf").exists() + assert (scans / "re.pdf").exists() and not (buha / "re.pdf").exists() + with session_factory() as s: + rule = s.get(Rule, rid) + assert rule is not None and rule.waiting_since is not None + assert "buha" in (rule.waiting_reason or "") + runs_before = len(rule.runs) + # weiterer Lauf während des Ausfalls: kein neuer Protokolleintrag + assert service.execute(rid) is None + with session_factory() as s: + rule = s.get(Rule, rid) + assert rule is not None and len(rule.runs) == runs_before + + fake_backend.mount_failures.clear() + with session_factory() as s: + buha_id = s.scalar(select(RuleTarget.mount_id).where(RuleTarget.name == "Buchhaltung")) + assert buha_id is not None + manager.mount(buha_id) # Anbindung wieder da + service.execute(rid) + assert (buha / "re.pdf").exists() and not (scans / "re.pdf").exists() + assert len(list(consume.iterdir())) == 1 # nicht doppelt geliefert + with session_factory() as s: + rule = s.get(Rule, rid) + assert rule is not None and rule.waiting_since is None + + +def test_all_targets_unavailable_makes_rule_wait( + setup: tuple[int, Path, Path, Path, Path], + manager: MountManager, + fake_backend: FakeMountBackend, + session_factory: sessionmaker[Session], +) -> None: + rid, scans, _consume, _arch, _buha = setup + with session_scope(session_factory) as s: + r = s.get(Rule, rid) + assert r is not None + r.dst_mount_id = r.targets[0].mount_id # Hauptziel ebenfalls auf „buha“ + r.dst_path = "dms" + r.targets = r.targets[:1] + write(scans / "doc.pdf") + fake_backend.vanish(manager.mountpoint_for("buha")) + fake_backend.fail_next_mount(times=50) + result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid) + assert result.unavailable and result.status is RunStatus.ERROR + assert (scans / "doc.pdf").exists() + + +def test_mount_used_as_target_cannot_be_deleted( + setup: tuple[int, Path, Path, Path, Path], + session_factory: sessionmaker[Session], + manager: MountManager, +) -> None: + from netmover.admin import AdminError, delete_mount + from netmover.models import Mount + + with session_scope(session_factory) as s: + arch_id = s.scalar(select(Mount.id).where(Mount.name == "archiv")) + assert arch_id is not None + with pytest.raises(AdminError, match="Wird noch von Regeln benutzt"): + delete_mount(s, manager, arch_id) + + +# -- Formular --------------------------------------------------------------------------------- + + +def _base(mid: int) -> dict[str, str]: + return { + "name": "x", + "src_mount_id": str(mid), + "dst_kind": "local", + "dst_path": "/srv/consume", + "interval_seconds": "30", + "stability_seconds": "5", + "op_timeout_seconds": "120", + } + + +def test_rule_form_target_validation( + session_factory: sessionmaker[Session], make_mount: MountFactory +) -> None: + from netmover.admin import parse_rule_form + + mid = make_mount() + base = _base(mid) + with session_scope(session_factory) as s: + _, err = parse_rule_form( + s, {**base, "target_0_kind": "local", "target_0_path": "/srv/consume"} + ) + assert "Hauptziel" in err["target_0_path"] + _, err = parse_rule_form(s, {**base, "target_0_kind": "mount", "target_0_path": "a"}) + assert "target_0_mount_id" in err + _, err = parse_rule_form(s, {**base, "target_0_kind": "local", "target_0_path": "rel"}) + assert "absoluten Pfad" in err["target_0_path"] + two = { + "target_0_kind": "local", + "target_0_path": "/srv/a", + "target_1_kind": "local", + "target_1_path": "/srv/a", + } + _, err = parse_rule_form(s, {**base, **two}) + assert "Derselbe Ordner" in err["target_1_path"] + _, err = parse_rule_form( + s, + { + **base, + **two, + "target_1_path": "/srv/b", + "target_0_name": "X", + "target_1_name": "X", + }, + ) + assert "schon vor" in err["target_1_name"] + _, err = parse_rule_form( + s, {**base, "target_0_kind": "mount", "target_0_mount_id": str(mid)} + ) + assert "derselbe Ordner" in err["target_0_path"] # = Quelle + + +def test_rule_form_saves_compacts_and_keeps_ledger( + session_factory: sessionmaker[Session], make_mount: MountFactory +) -> None: + from netmover.admin import parse_rule_form, save_rule + + mid = make_mount() + arch = make_mount("archiv", share="archiv") + form = { + **_base(mid), + # Platz 0 leer gelassen, Plätze 1 und 3 belegt → kompakt 0, 1 + "target_1_kind": "mount", + "target_1_mount_id": str(arch), + "target_1_path": "rechnungen", + "target_1_layout": "date", + "target_3_kind": "local", + "target_3_name": "Buchhaltung", + "target_3_path": "/srv/buha", + } + with session_scope(session_factory) as s: + d, err = parse_rule_form(s, form) + assert err == {} + r, _ = save_rule(s, d) + assert r is not None + rid = r.id + got = [(t.position, t.name, t.kind.value, t.path, t.layout.value) for t in r.targets] + assert got == [ + (0, "archiv", "mount", "rechnungen", "date"), # Name = Anbindung + (1, "Buchhaltung", "local", "/srv/buha", "flat"), + ] + first, second = r.targets[0].id, r.targets[1].id + for tid in (first, second): + s.add(TargetFile(target_id=tid, src_relpath="a", size=1, mtime_ns=1, dst_path="x")) + # erneut speichern: Archiv unverändert, Buchhaltung mit neuem Ordner + with session_scope(session_factory) as s: + d, err = parse_rule_form(s, {**form, "target_3_path": "/srv/buha2"}) + assert err == {} + save_rule(s, d, rid) + with session_factory() as s: + r = s.get(Rule, rid) + assert r is not None and [t.id for t in r.targets] == [first, second] + kept = {f.target_id for f in s.scalars(select(TargetFile))} + assert kept == {first} # Buchführung nur bei geändertem Ordner verworfen + # Ziel entfernen + with session_scope(session_factory) as s: + d, _ = parse_rule_form(s, {**form, "target_3_kind": ""}) + save_rule(s, d, rid) + with session_factory() as s: + r = s.get(Rule, rid) + assert r is not None and [t.name for t in r.targets] == ["archiv"] + + +def test_rule_form_page_has_nine_target_slots(client: Any, state: Any) -> None: + html = client.get("/rules/new").text + assert "Weitere Ziele" in html and 'name="target_8_kind"' in html + assert 'name="target_9_kind"' not in html + assert 'data-show-for="target_0_kind:mount local|target_1_kind:mount local"' in html + + +# -- Export / Import -------------------------------------------------------------------------- + + +def test_export_import_targets_and_legacy_archive( + setup: tuple[int, Path, Path, Path, Path], + session_factory: sessionmaker[Session], + manager: MountManager, +) -> None: + from netmover.settings_store import AppSettings + from netmover.transfer import export_config, import_config + + with session_factory() as s: + data = export_config(s, AppSettings()) + (rule,) = data["rules"] + assert [(t["name"], t["mount"], t["path"]) for t in rule["targets"]] == [ + ("Buchhaltung", "buha", ""), + ("Archiv", "archiv", "rechnungen"), + ] + new = {**rule, "name": "Kopie"} + legacy = {k: v for k, v in rule.items() if k != "targets"} | { + "name": "Alt", + "archive_kind": "mount", + "archive_mount": "archiv", + "archive_path": "alt", + "archive_layout": "date", + } + with session_scope(session_factory) as s: + report = import_config(s, manager, {**data, "mounts": [], "rules": [new, legacy]}) + assert report.errors == [] and len(report.created) == 2 + with session_factory() as s: + by_name = {r.name: r for r in s.scalars(select(Rule))} + assert [t.name for t in by_name["Kopie"].targets] == ["Buchhaltung", "Archiv"] + (t,) = by_name["Alt"].targets + assert (t.name, t.path, t.layout.value) == ("Archiv", "alt", "date")