diff --git a/CHANGELOG.md b/CHANGELOG.md index 786a7ca..4ab369f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,26 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`. wachsendem Abstand neu eingehängt. - Mountpoints ausschließlich unter `/mnt/netmover/`, müssen vor dem Einhängen leer sein und werden nach dem Aushängen entfernt. +- Mover-Engine: Stabilitätsprüfung (Größe **und** mtime nach dem + Stabilitätsfenster), unfertige/versteckte Dateien und Synology-Ordner + (`@eaDir`, `#recycle`) werden ignoriert, Glob (mehrere Muster mit `;`) oder Regex. +- Atomares Schreiben: `.netmover-.part` → `fsync` → Größenprüfung → + Metadaten → Quellprüfung → Umbenennen ohne Überschreiben (`link`, Rückfall auf + `rename`) → erst dann Quelle löschen; gleiches Gerät ohne Kopie per `link`. +- Namenskollisionen: umbenennen (`scan-1.pdf`, `scan-2.pdf`, …, jedes Mal gegen + das Dateisystem geprüft), überspringen oder überschreiben. +- Buchführung im Kopiermodus (`copied_files`); verhindert auch Duplikate, wenn im + Verschiebemodus die Quelle nicht gelöscht werden kann. +- Platzprüfung vor jeder Kopie (Größe + 5 %), Übernahme von mtime und Modus, + optional Eigentümer; Metadaten-Fehler sind Hinweise, keine Fehler; `chown` auf + CIFS und auf NFS mit Root-Squash wird übersprungen und einmalig gemeldet. +- Quarantäneordner für Dateien, die wiederholt scheitern. +- SFTP-Ziel mit demselben atomaren Ablauf (`rename` ohne Überschreiben, + `posix-rename` beim Überschreiben, `fsync`/`statvfs` über OpenSSH-Erweiterungen). +- Regellauf mit Lock pro Regel, Lebenszeichen der Anbindungen und Worker mit + Inaktivitäts-Zeitlimit; ein hängender Worker blockiert nur seine eigene Regel. +- Lokale Ziele unter dem Mount-Basisverzeichnis werden abgelehnt (dort muss die + Anbindung als Ziel gewählt werden, damit „eingehängt?“ geprüft wird). ### Behoben - Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den diff --git a/DEVLOG.md b/DEVLOG.md index 75f1368..760504c 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -9,10 +9,52 @@ Zeiten in Ortszeit (Europe/Berlin). |------------|-------|-------|-------|---------|----------------------------------------| | 2026-10-10 | 17:21 | 17:27 | 0:06 | 1 | Gerüst, Datenmodell, Migrationen | | 2026-10-10 | 18:35 | 18:48 | 0:13 | 2 | Mount-Schicht, Validierung, SFTP | -| **Summe** | | | **0:19** | | | +| 2026-10-10 | 18:53 | 19:07 | 0:14 | 3 | Mover-Logik | +| **Summe** | | | **0:33** | | | --- +## 2026-10-10 · 18:53–19:07 · Schritt 3: Mover-Logik + +**Erledigt** +- Remote `origin` (Gitea) war eingerichtet → `main` gepusht. +- `mover/scan.py`, `naming.py`, `ledger.py`, `destinations.py` (lokal/Kernel-Mount + und SFTP), `engine.py` (Ablauf pro Lauf), `runner.py` (Lock, Lebenszeichen, + Watchdog, Ziel-Aufbau). +- `timeouts.run_with_watchdog`: Inaktivitäts-Zeitlimit mit `CancelToken`. +- 60 neue Tests (183 gesamt), u. a. Abbruch in jeder Phase, Wettlauf um + Dateinamen, hängender Lauf, SFTP-Ziel mit Kollision gegen echten SFTP-Server. +- Mutationsprobe: 9 eingebaute Fehler (keine Stabilitätsprüfung, kein fsync, + keine Quellprüfung, Wettlauf, direkt unter Endnamen schreiben, keine + Buchführung, keine Platzprüfung, Lock-Freigabe, kein Watchdog). Zwei überlebten + zunächst (fsync, enges Wettlauf-Fenster) → zwei Tests ergänzt; jetzt alle erkannt. + +**Entscheidungen** +- Zeitlimit als **Inaktivitäts**-Watchdog statt Gesamtlaufzeit: große Dateien + über langsame Leitungen werden nicht abgebrochen, hängende Zugriffe schon. +- Regel-Lock gibt erst der Worker selbst frei. Nach einer Zeitüberschreitung + bleibt die Regel blockiert, bis der hängende Worker endet – so kopieren nie + zwei Läufe dieselbe Datei gleichzeitig. Statusseite zeigt „läuft seit …“. +- Stabilität innerhalb eines Laufs (zweite Messung nach dem Fenster) plus + Quellprüfung direkt vor dem Umbenennen. +- Umbenennen ohne Überschreiben per `link()` (atomar, EEXIST bei Kollision); + ohne Hardlink-Unterstützung (oft CIFS) Rückfall auf „prüfen + rename“. +- Bei Rekursion wird die Ordnerstruktur im Ziel beibehalten; Trennung nach + Quell-Unterordnern geht über mehrere Regeln mit eigenem Unterpfad. +- Quarantänepfad ist relativ zur Quell-Anbindung (gleiches Dateisystem → rename). +- `copied_files` wird nach jedem *vollständigen* Scan bereinigt (statt nur beim + Protokoll-Aufräumen) – zeitnäher, und nie nach einem abgebrochenen Scan. +- ESTALE/EIO/ENOTCONN mitten im Lauf → Lauf abbrechen (nicht jede Datei einzeln + scheitern lassen, nicht gegen den Quarantäne-Zähler zählen). + +**Stolpersteine** +- `tmp_path` liegt auf einem Gerät → Tests liefen erst über den Schnellweg + (`link`) statt über den Kopierweg; Fixture `cross_device` erzwingt ihn. +- Prüfung „lokales Ziel nicht unter Mount-Basis“ nutzte nur `resolve()` – ein + Symlink umgeht das. Jetzt wörtlich **und** aufgelöst. +- Lock-Übergabe an den Worker: doppelte Freigabe bei sehr schnellem Lauf + (beim Gegenlesen gefunden, vor dem ersten Testlauf behoben). + ## 2026-10-10 · 18:35–18:48 · Schritt 2: Mount-Schicht, Validierung, SFTP **Anforderungsänderung (Rücksprache)** diff --git a/README.md b/README.md index 60e479e..4290b73 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ Selbst gehosteter Dienst mit Web-Oberfläche, der SMB/CIFS- und NFS-Freigaben verwaltet und Dateien zeitgesteuert zwischen ihnen verschiebt. -> Status: in Entwicklung (Schritt 2 von 6). Vollständige Doku folgt mit Schritt 6. +> Status: in Entwicklung (Schritt 3 von 6). Vollständige Doku folgt mit Schritt 6. ## Entwicklung diff --git a/src/netmover/mounting/manager.py b/src/netmover/mounting/manager.py index 04e5cd8..50d4454 100644 --- a/src/netmover/mounting/manager.py +++ b/src/netmover/mounting/manager.py @@ -585,3 +585,10 @@ class MountManager: except (OSError, OperationTimeout): return None return st.f_blocks * st.f_frsize, st.f_bavail * st.f_frsize + + def sftp_config_for(self, mount_id: int) -> SftpConfig: + """Entschlüsselte Verbindungsdaten einer SFTP-Anbindung für einen Regellauf.""" + m, cred = self._load(mount_id) + if m.type is not MountType.SFTP: + raise SftpError(f"„{m.name}“ ist keine SFTP-Anbindung") + return sftp_config(MountConfig.from_model(m), cred) diff --git a/src/netmover/mover/runner.py b/src/netmover/mover/runner.py new file mode 100644 index 0000000..cb96dfa --- /dev/null +++ b/src/netmover/mover/runner.py @@ -0,0 +1,272 @@ +"""Einen Regellauf sicher ausführen. + +* Fallstrick 8: ein Lock pro Regel; läuft sie noch, wird der Termin übersprungen. +* Fallstrick 2: vor jedem Lauf prüft der :class:`MountManager`, dass Quelle und + Ziel wirklich eingehängt und lebendig sind – sonst wird nichts angefasst. +* Fallstrick 13: die Dateizugriffe laufen in einem eigenen Worker mit + Inaktivitäts-Zeitlimit. Der Lock wird erst freigegeben, wenn der Worker + wirklich endet – ein hängender Lauf kann also nie parallel zu einem neuen + dieselben Dateien bearbeiten; neue Termine werden bis dahin übersprungen. +""" + +from __future__ import annotations + +import posixpath +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path + +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.mounting.manager import MountManager, MountUnavailable +from netmover.mounting.sftp import SftpError, connect +from netmover.mover.destinations import ( + Destination, + LocalDestination, + SftpDestination, + TransferError, +) +from netmover.mover.engine import Engine, RuleSnapshot, RunResult +from netmover.mover.ledger import Ledger +from netmover.timeouts import CancelToken, OperationTimeout, run_with_watchdog +from netmover.validation import ValidationError, safe_abspath, safe_relpath + +log = get_logger(__name__) + + +@dataclass(frozen=True) +class RulePlan: + snapshot: RuleSnapshot + name: str + src_mount_id: int + src_type: MountType + src_subpath: str + dst_mount_id: int | None + dst_type: MountType | None + dst_name: str | None + dst_path: str + quarantine_path: str | None + op_timeout: int + dry_run: bool + + +def parse_owner(value: str | None) -> tuple[int, int] | None: + if not value: + return None + uid, _, gid = value.partition(":") + try: + return int(uid), int(gid or uid) + except ValueError: + raise ValidationError("set_owner", "Eigentümer bitte als „uid:gid“ angeben") from None + + +def parse_mode(value: str | None) -> int | None: + return int(value, 8) if value else None + + +def load_plan(session: Session, rule_id: int) -> RulePlan: + r = session.get(Rule, rule_id) + if r is None: + raise LookupError(f"Regel {rule_id} existiert nicht") + dst = r.dst_mount + snap = RuleSnapshot( + id=r.id, + name=r.name, + action=r.action, + on_collision=r.on_collision, + pattern=r.pattern, + pattern_type=r.pattern_type, + recursive=r.recursive, + stability_seconds=r.stability_seconds, + min_age_seconds=r.min_age_seconds, + max_size_mb=r.max_size_mb, + set_owner=parse_owner(r.set_owner), + 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, + ) + return RulePlan( + snapshot=snap, + name=r.name, + src_mount_id=r.src_mount_id, + src_type=r.src_mount.type, + src_subpath=safe_relpath("src_subpath", r.src_subpath), + dst_mount_id=r.dst_mount_id, + dst_type=dst.type if dst is not None else None, + dst_name=dst.name if dst is not None else None, + dst_path=r.dst_path, + quarantine_path=safe_relpath("quarantine_path", r.quarantine_path) + if r.quarantine_path + else None, + op_timeout=r.op_timeout_seconds, + dry_run=r.dry_run, + ) + + +class RuleRunner: + def __init__( + self, + session_factory: sessionmaker[Session], + mounts: MountManager, + *, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.time, + ) -> None: + self.session_factory = session_factory + self.mounts = mounts + self.engine = Engine(Ledger(session_factory), sleep=sleep, clock=clock) + self._clock = clock + self._locks: dict[int, threading.Lock] = {} + self._started: dict[int, float] = {} + self._guard = threading.Lock() + + def _lock(self, rule_id: int) -> threading.Lock: + with self._guard: + return self._locks.setdefault(rule_id, threading.Lock()) + + def running_since(self, rule_id: int) -> datetime | None: + ts = self._started.get(rule_id) + return datetime.fromtimestamp(ts, UTC) if ts is not None else None + + def run(self, rule_id: int, *, dry_run: bool | None = None) -> RunResult: + lock = self._lock(rule_id) + if not lock.acquire(blocking=False): + since = self.running_since(rule_id) + when = f" (seit {since.astimezone():%H:%M:%S})" if since else "" + return RunResult( + status=RunStatus.SKIPPED, + dry_run=bool(dry_run), + message=f"Termin übersprungen – voriger Lauf läuft noch{when}", + ) + # Wird True, sobald der Worker-Thread den Lock übernommen hat; ab dann gibt + # ihn ausschließlich der Worker frei (on_finish), auch nach einem Zeitlimit. + handed_over = [False] + try: + with session_scope(self.session_factory) as s: + plan = load_plan(s, rule_id) + dry = plan.dry_run if dry_run is None else dry_run + return self._run_locked(plan, dry, lock, rule_id, handed_over) + except (ValidationError, LookupError) as exc: + return RunResult(status=RunStatus.ERROR, dry_run=bool(dry_run), message=str(exc)) + finally: + if not handed_over[0]: + lock.release() + + def _run_locked( + self, + plan: RulePlan, + dry: bool, + lock: threading.Lock, + rule_id: int, + handed_over: list[bool], + ) -> RunResult: + if not plan.src_type.is_kernel_mount: + return RunResult( + RunStatus.ERROR, dry, message="SFTP-Anbindungen sind nur als Ziel möglich" + ) + 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 + ) + except MountUnavailable as exc: + return RunResult(RunStatus.ERROR, dry, message=str(exc)) + if src_mp is None: # nur bei SFTP, oben bereits ausgeschlossen + return RunResult(RunStatus.ERROR, dry, message="Quelle ist kein Kernel-Mount") + src_root = src_mp / plan.src_subpath if plan.src_subpath else src_mp + quarantine_root = src_mp / plan.quarantine_path if plan.quarantine_path else None + + try: + make_dest = self._dest_factory(plan, dst_mp, src_root) + except (ValidationError, TransferError, SftpError) as exc: + return RunResult(RunStatus.ERROR, dry, message=str(exc)) + + def work(token: CancelToken) -> RunResult: + dest = make_dest() + try: + return self.engine.execute( + plan.snapshot, + src_root, + dest, + token, + dry_run=dry, + quarantine_root=quarantine_root, + ) + finally: + dest.close() + + def finish() -> None: + self._started.pop(rule_id, None) + lock.release() + + self._started[rule_id] = self._clock() + handed_over[0] = True + try: + return run_with_watchdog(work, plan.op_timeout, f"rule:{plan.name}", on_finish=finish) + except OperationTimeout: + return RunResult( + RunStatus.ERROR, + dry, + message=( + f"Keine Reaktion seit {plan.op_timeout} s – Lauf abgebrochen. Vermutlich hängt " + "eine Anbindung; sie wird vor dem nächsten Lauf geprüft und ggf. neu " + "eingehängt." + ), + ) + except (TransferError, SftpError, OSError) as exc: + return RunResult(RunStatus.ERROR, dry, message=str(exc)) + + def _under_mount_base(self, path: Path) -> bool: + # Wörtlich und aufgelöst prüfen – ein Symlink darf die Prüfung nicht umgehen. + bases = {self.mounts.mount_base.absolute(), self.mounts.mount_base.resolve()} + paths = {path.absolute(), path.resolve()} + return any(p == b or b in p.parents for p in paths for b in bases) + + def _dest_factory( + self, plan: RulePlan, dst_mp: Path | None, src_root: Path + ) -> Callable[[], Destination]: + if plan.dst_mount_id is None: + root = Path(safe_abspath("dst_path", plan.dst_path)) + if self._under_mount_base(root): + base = self.mounts.mount_base + raise ValidationError( + "dst_path", + f"Lokales Ziel liegt unter {base} – bitte die Anbindung als Ziel wählen, " + "damit geprüft wird, ob sie eingehängt ist.", + ) + kind = "local" + local_root = root + create = False + elif plan.dst_type is MountType.SFTP: + scfg = self.mounts.sftp_config_for(plan.dst_mount_id) + sub = safe_relpath("dst_path", plan.dst_path) + remote = posixpath.join(scfg.base_path, sub) if sub else scfg.base_path + label = plan.dst_name or "sftp" + + def make_sftp() -> Destination: + return SftpDestination(connect(scfg), remote, label=label) + + return make_sftp + else: + if dst_mp is None: + raise TransferError("Ziel-Anbindung ist nicht eingehängt") + sub = safe_relpath("dst_path", plan.dst_path) + local_root = dst_mp / sub if sub else dst_mp + kind = "cifs" if plan.dst_type is MountType.CIFS else "nfs" + create = True + + if local_root.resolve() == src_root.resolve(): + raise ValidationError("dst_path", "Quelle und Ziel sind derselbe Ordner") + + def make_local() -> Destination: + return LocalDestination(local_root, kind, create_root=create) # type: ignore[arg-type] + + return make_local diff --git a/tests/test_runner.py b/tests/test_runner.py new file mode 100644 index 0000000..b08c903 --- /dev/null +++ b/tests/test_runner.py @@ -0,0 +1,307 @@ +"""Regellauf mit FakeMountBackend – Fallstricke 2, 8 und 13 sowie SFTP-Ziel.""" + +from __future__ import annotations + +import threading +import time +from collections.abc import Iterator +from pathlib import Path + +import pytest +from sqlalchemy.orm import Session, sessionmaker + +from netmover.crypto import SecretBox +from netmover.db import session_scope +from netmover.models import ( + Credential, + CredentialKind, + ItemStatus, + Mount, + MountType, + Rule, + RunStatus, +) +from netmover.mounting.fake import FakeMountBackend +from netmover.mounting.manager import MountManager +from netmover.mover.runner import RuleRunner +from tests.conftest import MountFactory +from tests.sftp_stub import StubSFTPServer + + +@pytest.fixture +def setup( + manager: MountManager, + fake_backend: FakeMountBackend, + make_mount: MountFactory, + session_factory: sessionmaker[Session], +) -> tuple[int, int, int, Path, Path]: + scans = fake_backend.export("nas", "scans") + archiv = fake_backend.export("nas2", "archiv") + src_id = make_mount("scans", server="nas", share="scans") + dst_id = make_mount("archiv", server="nas2", share="archiv") + manager.mount(src_id) + manager.mount(dst_id) + with session_scope(session_factory) as s: + r = Rule( + name="scans→archiv", + src_mount_id=src_id, + dst_mount_id=dst_id, + dst_path="eingang", + stability_seconds=1, + ) + s.add(r) + s.flush() + rid = r.id + return rid, src_id, dst_id, scans, archiv + + +@pytest.fixture +def runner(session_factory: sessionmaker[Session], manager: MountManager) -> RuleRunner: + return RuleRunner(session_factory, manager, sleep=lambda _: None) + + +def test_moves_between_two_mounts( + runner: RuleRunner, setup: tuple[int, int, int, Path, Path] +) -> None: + rid, _, _, scans, archiv = setup + (scans / "scan001.pdf").write_text("x") + result = runner.run(rid) + assert result.status is RunStatus.OK + assert [(i.src, i.status) for i in result.items] == [("scan001.pdf", ItemStatus.MOVED)] + assert (archiv / "eingang" / "scan001.pdf").read_text() == "x" + assert not (scans / "scan001.pdf").exists() + + +# -- Fallstrick 2 --------------------------------------------------------------------------- + + +def test_unmounted_destination_aborts_and_writes_nothing_locally( + runner: RuleRunner, + setup: tuple[int, int, int, Path, Path], + fake_backend: FakeMountBackend, + manager: MountManager, +) -> None: + rid, _, _, scans, _archiv = setup + (scans / "scan001.pdf").write_text("x") + mp = manager.mountpoint_for("archiv") + fake_backend.vanish(mp) # Freigabe weg, Mountpoint ist jetzt ein leeres lokales Verzeichnis + result = runner.run(rid) + assert result.status is RunStatus.ERROR + assert "nicht eingehängt" in (result.message or "") + assert list(mp.iterdir()) == [] # nichts auf die lokale Platte geschrieben + assert (scans / "scan001.pdf").exists() + + +def test_unmounted_source_aborts( + runner: RuleRunner, + setup: tuple[int, int, int, Path, Path], + fake_backend: FakeMountBackend, + manager: MountManager, +) -> None: + rid, *_ = setup + fake_backend.vanish(manager.mountpoint_for("scans")) + result = runner.run(rid) + assert result.status is RunStatus.ERROR and result.items == [] + + +def test_local_destination_under_mount_base_is_refused( + runner: RuleRunner, + setup: tuple[int, int, int, Path, Path], + manager: MountManager, + session_factory: sessionmaker[Session], +) -> None: + rid, *_ = setup + with session_scope(session_factory) as s: + r = s.get(Rule, rid) + assert r is not None + r.dst_mount_id = None + r.dst_path = str(manager.mountpoint_for("archiv")) + result = runner.run(rid) + assert result.status is RunStatus.ERROR + assert "Anbindung als Ziel" in (result.message or "") + + +def test_local_destination( + runner: RuleRunner, + setup: tuple[int, int, int, Path, Path], + tmp_path: Path, + session_factory: sessionmaker[Session], +) -> None: + rid, _, _, scans, _ = setup + consume = tmp_path / "consume" + consume.mkdir() + with session_scope(session_factory) as s: + r = s.get(Rule, rid) + assert r is not None + r.dst_mount_id = None + r.dst_path = str(consume) + (scans / "a.pdf").write_text("x") + assert runner.run(rid).status is RunStatus.OK + assert (consume / "a.pdf").exists() + + +# -- Fallstrick 8 / 13: Überlappung und Zeitlimit --------------------------------------------- + + +@pytest.fixture +def blocking_runner( + session_factory: sessionmaker[Session], manager: MountManager +) -> Iterator[tuple[RuleRunner, threading.Event, threading.Event]]: + entered = threading.Event() + release = threading.Event() + + def sleep(_: float) -> None: # Stabilitätsfenster „hängt“ + entered.set() + release.wait(10) + + yield RuleRunner(session_factory, manager, sleep=sleep), entered, release + release.set() + + +def test_overlapping_run_is_skipped( + blocking_runner: tuple[RuleRunner, threading.Event, threading.Event], + setup: tuple[int, int, int, Path, Path], +) -> None: + runner, entered, release = blocking_runner + rid, _, _, scans, archiv = setup + (scans / "a.pdf").write_text("x") + results = [] + t = threading.Thread(target=lambda: results.append(runner.run(rid))) + t.start() + assert entered.wait(5) + skipped = runner.run(rid) + assert skipped.status is RunStatus.SKIPPED + assert "läuft noch" in (skipped.message or "") + release.set() + t.join(5) + assert results[0].status is RunStatus.OK + assert (archiv / "eingang" / "a.pdf").exists() + + +def test_hanging_run_times_out_and_blocks_rule_until_worker_ends( + blocking_runner: tuple[RuleRunner, threading.Event, threading.Event], + setup: tuple[int, int, int, Path, Path], + session_factory: sessionmaker[Session], +) -> None: + runner, _entered, release = blocking_runner + rid, _, _, scans, archiv = setup + with session_scope(session_factory) as s: + r = s.get(Rule, rid) + assert r is not None + r.op_timeout_seconds = 1 + (scans / "a.pdf").write_text("x") + + t0 = time.monotonic() + result = runner.run(rid) + assert time.monotonic() - t0 < 5 # Scheduler wird nicht blockiert + assert result.status is RunStatus.ERROR + assert "Keine Reaktion" in (result.message or "") + assert runner.running_since(rid) is not None + + # solange der alte Worker hängt: neue Termine überspringen + assert runner.run(rid).status is RunStatus.SKIPPED + + # Worker wacht auf → sieht den Abbruch, verschiebt nichts mehr, gibt Lock frei + release.set() + for _ in range(50): + if runner.running_since(rid) is None: + break + time.sleep(0.05) + assert runner.running_since(rid) is None + assert (scans / "a.pdf").exists() + assert not (archiv / "eingang" / "a.pdf").exists() + + +def test_stale_source_is_recovered_before_run( + runner: RuleRunner, + setup: tuple[int, int, int, Path, Path], + fake_backend: FakeMountBackend, + manager: MountManager, +) -> None: + import errno + + rid, _, _, scans, archiv = setup + (scans / "a.pdf").write_text("x") + fake_backend.inject_fault(manager.mountpoint_for("scans"), errno.ESTALE) + result = runner.run(rid) + assert result.status is RunStatus.OK + assert (archiv / "eingang" / "a.pdf").exists() + assert ("unmount", True) in [(c.action, c.lazy) for c in fake_backend.calls] + + +def test_dry_run_flag_from_rule_and_override( + runner: RuleRunner, + setup: tuple[int, int, int, Path, Path], + session_factory: sessionmaker[Session], +) -> None: + rid, _, _, scans, _archiv = setup + (scans / "a.pdf").write_text("x") + with session_scope(session_factory) as s: + r = s.get(Rule, rid) + assert r is not None + r.dry_run = True + result = runner.run(rid) + assert result.dry_run and result.items[0].status is ItemStatus.PLANNED + assert (scans / "a.pdf").exists() + assert runner.run(rid, dry_run=False).items[0].status is ItemStatus.MOVED + + +def test_unknown_rule(runner: RuleRunner) -> None: + assert runner.run(999).status is RunStatus.ERROR + + +# -- SFTP-Ziel -------------------------------------------------------------------------------- + + +@pytest.fixture +def sftp_server(tmp_path: Path) -> Iterator[StubSFTPServer]: + root = tmp_path / "sftp-root" + (root / "consume").mkdir(parents=True) + srv = StubSFTPServer(root).start() + yield srv + srv.stop() + + +def test_move_to_sftp_destination_with_collision( + runner: RuleRunner, + setup: tuple[int, int, int, Path, Path], + sftp_server: StubSFTPServer, + session_factory: sessionmaker[Session], + secretbox: SecretBox, +) -> None: + rid, _, _, scans, _ = setup + with session_scope(session_factory) as s: + c = Credential( + name="pk", + kind=CredentialKind.SSH_KEY, + username=sftp_server.username, + secret_enc=secretbox.encrypt(sftp_server.client_private_key), + ) + s.add(c) + s.flush() + m = Mount( + name="paperless", + slug="paperless", + type=MountType.SFTP, + server="127.0.0.1", + share="/consume", + port=sftp_server.port, + cred_id=c.id, + host_key=sftp_server.host_key_string, + ) + s.add(m) + s.flush() + r = s.get(Rule, rid) + assert r is not None + r.dst_mount_id = m.id + r.dst_path = "" + consume = sftp_server.root / "consume" + (consume / "scan.pdf").write_text("alt") + (scans / "scan.pdf").write_bytes(b"%PDF neu" * 50_000) + result = runner.run(rid) + assert result.status is RunStatus.OK, result.message + assert result.items[0].dst == "paperless:/consume/scan-1.pdf" + assert (consume / "scan-1.pdf").read_bytes() == b"%PDF neu" * 50_000 + assert (consume / "scan.pdf").read_text() == "alt" + assert not (scans / "scan.pdf").exists() + assert sorted(p.name for p in consume.iterdir()) == ["scan-1.pdf", "scan.pdf"]