From 5633d3057a5603bdc1a52ad90fa79fe40aa676d0 Mon Sep 17 00:00:00 2001 From: Florian Wilde Date: Sat, 10 Oct 2026 19:06:14 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Mover-Engine=20mit=20atomarem=20Schreib?= =?UTF-8?q?en,=20Stabilit=C3=A4t=20und=20Quarant=C3=A4ne?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lokale/Kernel-Mount- und SFTP-Ziele schreiben in .part, fsyncen, prüfen Größe und Quelle und benennen ohne Überschreiben um; erst dann wird die Quelle gelöscht. Stabilitätsfenster, Platzprüfung, Metadaten, Berechtigungen, Kopiermodus mit Gedächtnis, Quarantäne. Co-Authored-By: Claude Opus 5.5 --- src/netmover/mover/destinations.py | 489 +++++++++++++++++++++ src/netmover/mover/engine.py | 434 ++++++++++++++++++ tests/test_engine.py | 680 +++++++++++++++++++++++++++++ 3 files changed, 1603 insertions(+) create mode 100644 src/netmover/mover/destinations.py create mode 100644 src/netmover/mover/engine.py create mode 100644 tests/test_engine.py diff --git a/src/netmover/mover/destinations.py b/src/netmover/mover/destinations.py new file mode 100644 index 0000000..2fddcc3 --- /dev/null +++ b/src/netmover/mover/destinations.py @@ -0,0 +1,489 @@ +"""Ziele: lokales Verzeichnis bzw. Kernel-Mount, oder SFTP. + +Fallstrick 3 – nie eine wachsende Datei unter ihrem Endnamen, nie beide Kopien +verlieren. Ablauf für jede Datei: + +1. in ``/.netmover-.part`` schreiben, ``flush`` + ``fsync``, + Größe prüfen, +2. Metadaten setzen (mtime, Modus, Eigentümer – Fehler sind nur Hinweise), +3. prüfen, dass sich die Quelle seit der Stabilitätsmessung nicht verändert hat, +4. atomar auf den Endnamen umbenennen – ohne zu überschreiben (``link`` bzw. + SFTP-``rename``), außer „überschreiben“ ist ausdrücklich eingestellt, +5. erst danach löscht der Aufrufer die Quelle. + +Scheitert irgendein Schritt vor 4, wird nur die ``.part``-Datei entfernt; die +Quelle ist unangetastet. Bricht der Prozess hart ab, bleibt höchstens eine +``.part``-Datei zurück, die :meth:`Destination.prepare` später aufräumt. +""" + +from __future__ import annotations + +import contextlib +import errno +import os +import posixpath +import stat +import time +import uuid +from abc import ABC, abstractmethod +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Literal + +from netmover.models import CollisionPolicy +from netmover.mounting.sftp import SftpSession +from netmover.mover.naming import collision_candidates +from netmover.timeouts import CancelToken + +CHUNK = 1024 * 1024 +STALE_PART_SECONDS = 3600 +PART_PREFIX = ".netmover-" +PART_SUFFIX = ".part" + +FsKind = Literal["local", "cifs", "nfs", "sftp"] + +# Fehler, bei denen ``link`` nicht geht und wir auf „prüfen + rename“ ausweichen. +_NO_LINK_ERRNOS = frozenset( + {errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EMLINK, errno.EXDEV, errno.ENOSYS} +) + + +class TransferError(Exception): + """Übertragung dieser Datei gescheitert (Meldung im Klartext).""" + + +class SourceChanged(TransferError): + """Quelle hat sich seit der Stabilitätsmessung verändert – später erneut versuchen.""" + + +class CollisionSkip(Exception): + """Ziel existiert und die Regel sagt „überspringen“.""" + + +@dataclass(frozen=True) +class MetaOptions: + atime_ns: int + mtime_ns: int + mode: int | None # Rechte-Bits, None = nicht setzen + owner: tuple[int, int] | None # None = nicht setzen + + +@dataclass +class CommitResult: + final_rel: str + notes: list[str] = field(default_factory=list) + source_consumed: bool = False # True: Quelle wurde per rename bereits „verschoben“ + + +Verify = Callable[[], None] + + +class Destination(ABC): + kind: FsKind + + @abstractmethod + def describe(self, rel: str) -> str: + """Lesbarer Zielpfad für Protokoll und Probelauf.""" + + @abstractmethod + def prepare(self) -> None: + """Vor dem Lauf: Zielordner prüfen, alte ``.part``-Reste entfernen.""" + + @abstractmethod + def exists(self, rel: str) -> bool: ... + + @abstractmethod + def free_bytes(self) -> int | None: ... + + def close(self) -> None: # noqa: B027 – absichtlich optional + pass + + # -- Primitive der Unterklassen ------------------------------------------------- + + @abstractmethod + def _ensure_parent(self, rel: str) -> None: ... + + @abstractmethod + def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]: ... + + @abstractmethod + def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]: ... + + @abstractmethod + def _link_noreplace(self, tmp: str, final: str) -> None: + """Atomar umbenennen; ``FileExistsError``, wenn ``final`` existiert.""" + + @abstractmethod + def _replace(self, tmp: str, final: str) -> list[str]: ... + + @abstractmethod + def _remove(self, rel: str) -> None: ... + + def _sync_dir(self, rel_dir: str) -> None: # noqa: B027 + pass + + # -- gemeinsamer Ablauf --------------------------------------------------------- + + def plan(self, rel: str, policy: CollisionPolicy) -> str | None: + """Voraussichtlicher Endname (Probelauf / Vorabprüfung); None = überspringen.""" + if policy is CollisionPolicy.OVERWRITE: + return rel + if policy is CollisionPolicy.SKIP: + return None if self.exists(rel) else rel + for cand in collision_candidates(rel): + if not self.exists(cand): + return cand + raise TransferError(f"Kein freier Dateiname für {rel}") + + def commit( + self, + src: Path, + size: int, + rel: str, + policy: CollisionPolicy, + meta: MetaOptions, + verify: Verify, + token: CancelToken, + ) -> CommitResult: + self._ensure_parent(rel) + tmp = posixpath.join( + posixpath.dirname(rel), f"{PART_PREFIX}{uuid.uuid4().hex}{PART_SUFFIX}" + ) + try: + notes = self._upload(src, tmp, size, token) + notes += self._apply_meta(tmp, meta) + verify() + token.check() + final, more = self._finalize(tmp, rel, policy) + notes += more + except BaseException: + with contextlib.suppress(Exception): + self._remove(tmp) + raise + self._sync_dir(posixpath.dirname(final)) + return CommitResult(final, notes) + + def move_same_device( + self, src: Path, rel: str, policy: CollisionPolicy, meta: MetaOptions + ) -> CommitResult | None: + """Schneller Weg ohne Kopie, wenn Quelle und Ziel auf demselben Gerät liegen.""" + return None + + def _finalize(self, tmp: str, rel: str, policy: CollisionPolicy) -> tuple[str, list[str]]: + if policy is CollisionPolicy.OVERWRITE: + return rel, self._replace(tmp, rel) + candidates = collision_candidates(rel) if policy is CollisionPolicy.RENAME else iter([rel]) + for cand in candidates: + if self.exists(cand): + if policy is CollisionPolicy.SKIP: + raise CollisionSkip + continue + try: + self._link_noreplace(tmp, cand) + except FileExistsError: + # Wettlauf: jemand anderes hat den Namen gerade belegt + if policy is CollisionPolicy.SKIP: + raise CollisionSkip from None + continue + return cand, [] + raise TransferError(f"Kein freier Dateiname für {rel}") + + +class LocalDestination(Destination): + """Lokales Verzeichnis oder Verzeichnis auf einem (geprüften) Kernel-Mount.""" + + def __init__( + self, + root: Path, + kind: Literal["local", "cifs", "nfs"] = "local", + *, + create_root: bool = False, + ) -> None: + self.root = root + self.kind = kind + self.create_root = create_root + + def _p(self, rel: str) -> Path: + return self.root / rel + + def describe(self, rel: str) -> str: + return str(self._p(rel)) + + def prepare(self) -> None: + if not self.root.is_dir(): + if not self.create_root: + raise TransferError(f"Zielordner {self.root} existiert nicht") + self.root.mkdir(mode=0o755, parents=True, exist_ok=True) + cutoff = time.time() - STALE_PART_SECONDS + for entry in os.scandir(self.root): + if entry.name.startswith(PART_PREFIX) and entry.name.endswith(PART_SUFFIX): + with contextlib.suppress(OSError): + if entry.stat(follow_symlinks=False).st_mtime < cutoff: + os.unlink(entry.path) + + def exists(self, rel: str) -> bool: + return os.path.lexists(self._p(rel)) + + def free_bytes(self) -> int | None: + st = os.statvfs(self.root) + return st.f_bavail * st.f_frsize + + def _ensure_parent(self, rel: str) -> None: + self._p(rel).parent.mkdir(mode=0o755, parents=True, exist_ok=True) + + def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]: + written = 0 + fd = os.open(self._p(tmp), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with src.open("rb") as fin: + while chunk := fin.read(CHUNK): + token.check() + view = memoryview(chunk) + while view: + n = os.write(fd, view) + view = view[n:] + written += len(chunk) + if written != size: + raise SourceChanged( + "Quelle hat sich während der Übertragung geändert " + f"({written} statt {size} Bytes)" + ) + os.fsync(fd) + if os.fstat(fd).st_size != size: + raise TransferError("Größe der Zieldatei stimmt nach dem Schreiben nicht") + finally: + os.close(fd) + return [] + + def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]: + notes: list[str] = [] + path = self._p(rel) + if self.kind != "cifs": # auf CIFS bestimmen die Mount-Optionen Eigentümer/Rechte + if meta.owner is not None: + try: + os.chown(path, *meta.owner) + except PermissionError: + notes.append( + f"Eigentümer {meta.owner[0]}:{meta.owner[1]} nicht gesetzt – keine " + "Berechtigung (läuft NetMover als root? Root-Squash?)" + ) + except OSError as exc: + notes.append(f"Eigentümer nicht gesetzt: {exc.strerror}") + if meta.mode is not None: + try: + os.chmod(path, meta.mode) + except OSError as exc: + notes.append(f"Modus {meta.mode:04o} nicht gesetzt: {exc.strerror}") + try: + os.utime(path, ns=(meta.atime_ns, meta.mtime_ns)) + except OSError as exc: + notes.append(f"Änderungszeit nicht übernommen: {exc.strerror}") + return notes + + def _link_noreplace(self, tmp: str, final: str) -> None: + src, dst = self._p(tmp), self._p(final) + try: + os.link(src, dst) + except FileExistsError: + raise + except OSError as exc: + if exc.errno not in _NO_LINK_ERRNOS: + raise + # Dateisystem ohne Hardlinks (häufig CIFS): prüfen + rename. + # Kleines Wettlauf-Fenster, aber nie eine halbe Datei unter dem Endnamen. + if os.path.lexists(dst): + raise FileExistsError(errno.EEXIST, "exists", str(dst)) from None + os.rename(src, dst) + return + with contextlib.suppress(OSError): + os.unlink(src) # Rest wird sonst später als alte .part-Datei aufgeräumt + + def _replace(self, tmp: str, final: str) -> list[str]: + os.replace(self._p(tmp), self._p(final)) + return [] + + def _remove(self, rel: str) -> None: + with contextlib.suppress(FileNotFoundError): + os.unlink(self._p(rel)) + + def _sync_dir(self, rel_dir: str) -> None: + with contextlib.suppress(OSError): + fd = os.open(self._p(rel_dir) if rel_dir else self.root, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(fd) + finally: + os.close(fd) + + def move_same_device( + self, src: Path, rel: str, policy: CollisionPolicy, meta: MetaOptions + ) -> CommitResult | None: + try: + same = os.stat(src).st_dev == os.stat(self.root).st_dev + except OSError: + return None + if not same: + return None + self._ensure_parent(rel) + consumed = True + if policy is CollisionPolicy.OVERWRITE: + os.replace(src, self._p(rel)) + final = rel + else: + final = "" + candidates = ( + collision_candidates(rel) if policy is CollisionPolicy.RENAME else iter([rel]) + ) + for cand in candidates: + if self.exists(cand): + if policy is CollisionPolicy.SKIP: + raise CollisionSkip + continue + try: + os.link(src, self._p(cand)) + except FileExistsError: + if policy is CollisionPolicy.SKIP: + raise CollisionSkip from None + continue + except OSError as exc: + if exc.errno not in _NO_LINK_ERRNOS: + raise + if self.exists(cand): + continue + os.rename(src, self._p(cand)) + final = cand + break + # Ziel ist vollständig da; den alten Namen entfernt der Aufrufer – + # mit derselben Fehlerbehandlung wie nach einer Kopie. + consumed = False + final = cand + break + if not final: + raise TransferError(f"Kein freier Dateiname für {rel}") + notes = self._apply_meta(final, meta) + self._sync_dir(posixpath.dirname(final)) + return CommitResult(final, notes, source_consumed=consumed) + + +class SftpDestination(Destination): + kind: FsKind = "sftp" + + def __init__(self, session: SftpSession, base: str, label: str = "") -> None: + self.session = session + self.client = session.client + self.base = base + self.label = label + + def _p(self, rel: str) -> str: + return posixpath.join(self.base, rel) if rel else self.base + + def describe(self, rel: str) -> str: + prefix = f"{self.label}:" if self.label else "sftp:" + return prefix + self._p(rel) + + def close(self) -> None: + self.session.close() + + def prepare(self) -> None: + try: + attrs = self.client.stat(self.base) + except OSError as exc: + raise TransferError(f"SFTP-Zielordner {self.base}: {exc}") from None + if attrs.st_mode is None or not stat.S_ISDIR(attrs.st_mode): + raise TransferError(f"SFTP-Ziel {self.base} ist kein Verzeichnis") + cutoff = time.time() - STALE_PART_SECONDS + for a in self.client.listdir_attr(self.base): + is_part = a.filename.startswith(PART_PREFIX) and a.filename.endswith(PART_SUFFIX) + if is_part and a.st_mtime is not None and a.st_mtime < cutoff: + with contextlib.suppress(OSError): + self.client.remove(self._p(a.filename)) + + def exists(self, rel: str) -> bool: + try: + self.client.lstat(self._p(rel)) + except FileNotFoundError: + return False + return True + + def free_bytes(self) -> int | None: + usage = self.session.statvfs(self.base) + return usage[1] if usage else None + + def _ensure_parent(self, rel: str) -> None: + parent = posixpath.dirname(rel) + if not parent: + return + current = self.base + for part in parent.split("/"): + current = posixpath.join(current, part) + try: + self.client.stat(current) + except FileNotFoundError: + self.client.mkdir(current, mode=0o755) + + def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]: + notes: list[str] = [] + written = 0 + with self.client.open(self._p(tmp), "wb") as fh, src.open("rb") as fin: + fh.set_pipelined(True) + while chunk := fin.read(CHUNK // 4): + token.check() + fh.write(chunk) + written += len(chunk) + fh.flush() + if written != size: + raise SourceChanged( + "Quelle hat sich während der Übertragung geändert " + f"({written} statt {size} Bytes)" + ) + if not self.session.fsync(fh): + notes.append("SFTP-Server unterstützt kein fsync") + remote = self.client.stat(self._p(tmp)).st_size + if remote != size: + raise TransferError(f"Zieldatei hat {remote} statt {size} Bytes") + return notes + + def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]: + notes: list[str] = [] + path = self._p(rel) + if meta.owner is not None: + try: + self.client.chown(path, *meta.owner) + except OSError as exc: + notes.append(f"Eigentümer per SFTP nicht gesetzt: {exc}") + if meta.mode is not None: + try: + self.client.chmod(path, meta.mode) + except OSError as exc: + notes.append(f"Modus per SFTP nicht gesetzt: {exc}") + try: + self.client.utime(path, (meta.atime_ns / 1e9, meta.mtime_ns / 1e9)) + except OSError as exc: + notes.append(f"Änderungszeit nicht übernommen: {exc}") + return notes + + def _link_noreplace(self, tmp: str, final: str) -> None: + if self.exists(final): + raise FileExistsError(errno.EEXIST, "exists", final) + try: + # SFTPv3-rename überschreibt nicht (OpenSSH: link + unlink). + self.client.rename(self._p(tmp), self._p(final)) + except OSError: + if self.exists(final): + raise FileExistsError(errno.EEXIST, "exists", final) from None + raise + + def _replace(self, tmp: str, final: str) -> list[str]: + try: + self.client.posix_rename(self._p(tmp), self._p(final)) + except OSError: + # Server ohne posix-rename: löschen + umbenennen (kurz kein Ziel, nie halb) + with contextlib.suppress(FileNotFoundError): + self.client.remove(self._p(final)) + self.client.rename(self._p(tmp), self._p(final)) + return ["Server kann nicht atomar überschreiben – Ziel kurz nicht vorhanden"] + return [] + + def _remove(self, rel: str) -> None: + with contextlib.suppress(FileNotFoundError): + self.client.remove(self._p(rel)) diff --git a/src/netmover/mover/engine.py b/src/netmover/mover/engine.py new file mode 100644 index 0000000..1213899 --- /dev/null +++ b/src/netmover/mover/engine.py @@ -0,0 +1,434 @@ +"""Ablauf eines Regellaufs – ohne Scheduler, ohne Datenbank-Protokoll. + +Reihenfolge pro Lauf: + +1. Ziel vorbereiten, Quelle scannen (Filter, versteckte/unfertige Dateien raus), +2. Kopier-Buchführung: bereits übertragene Dateiversionen still auslassen, +3. Größe und Mindestalter prüfen, +4. Stabilität: nach dem Stabilitätsfenster erneut messen – nur Dateien mit + 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. +""" + +from __future__ import annotations + +import errno +import os +import stat +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path + +from netmover.logging_setup import event, get_logger +from netmover.models import ( + CollisionPolicy, + ItemStatus, + PatternType, + RuleAction, + RunStatus, +) +from netmover.mover.destinations import ( + CollisionSkip, + Destination, + LocalDestination, + MetaOptions, + SourceChanged, + TransferError, +) +from netmover.mover.ledger import FailureInfo, Ledger +from netmover.mover.naming import collision_candidates +from netmover.mover.scan import Candidate, FileState, Matcher, current_state, scan +from netmover.timeouts import Cancelled, CancelToken + +log = get_logger(__name__) + +SPACE_RESERVE = 1.05 +# Fehler, die auf eine gestörte Anbindung hindeuten: Lauf abbrechen statt jede +# weitere Datei scheitern zu lassen – der nächste Lauf prüft und hängt neu ein. +MOUNT_ERRNOS = frozenset({errno.ESTALE, errno.EIO, errno.ENOTCONN, errno.EHOSTDOWN}) + + +class RunAborted(Exception): + pass + + +@dataclass(frozen=True) +class RuleSnapshot: + id: int + name: str + action: RuleAction = RuleAction.MOVE + on_collision: CollisionPolicy = CollisionPolicy.RENAME + pattern: str = "*" + pattern_type: PatternType = PatternType.GLOB + recursive: bool = False + stability_seconds: int = 5 + min_age_seconds: int = 0 + max_size_mb: int | None = None + set_owner: tuple[int, int] | None = None + set_mode: int | None = None + quarantine_after: int = 3 + dst_root_squash: bool | None = None # nur NFS-Ziele: Ergebnis des Schreibtests + + +@dataclass +class ItemResult: + src: str + status: ItemStatus + dst: str | None = None + size: int | None = None + message: str | None = None + + +@dataclass +class RunResult: + status: RunStatus = RunStatus.OK + dry_run: bool = False + seen: int = 0 + moved: int = 0 + skipped: int = 0 + failed: int = 0 + message: str | None = None + items: list[ItemResult] = field(default_factory=list) + + def add(self, item: ItemResult) -> None: + self.items.append(item) + if item.status in (ItemStatus.MOVED, ItemStatus.COPIED, ItemStatus.PLANNED): + self.moved += 1 + elif item.status in (ItemStatus.SKIPPED, ItemStatus.QUARANTINED): + self.skipped += 1 + else: + self.failed += 1 + + +def _describe_os_error(exc: OSError) -> str: + where = f" ({exc.filename})" if exc.filename else "" + return f"{exc.strerror or exc}{where}" + + +def _mb(n: int) -> str: + return f"{n / 1024 / 1024:.1f} MB" + + +class Engine: + def __init__( + self, + ledger: Ledger, + *, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.time, + warn_once: Callable[[str], bool] | None = None, + ) -> None: + self.ledger = ledger + self._sleep = sleep + self._clock = clock + self._warned: set[str] = set() + self._warn_once = warn_once or self._default_warn_once + + def _default_warn_once(self, key: str) -> bool: + if key in self._warned: + return False + self._warned.add(key) + return True + + # -- Lauf ------------------------------------------------------------------------- + + def execute( + self, + rule: RuleSnapshot, + src_root: Path, + dest: Destination, + token: CancelToken, + *, + dry_run: bool = False, + quarantine_root: Path | None = None, + ) -> RunResult: + result = RunResult(dry_run=dry_run) + try: + self._execute(rule, src_root, dest, token, result, dry_run, quarantine_root) + except Cancelled: + result.status = RunStatus.ERROR + result.message = "Lauf abgebrochen" + except RunAborted as exc: + result.status = RunStatus.ERROR + result.message = str(exc) + except TransferError as exc: + result.status = RunStatus.ERROR + result.message = str(exc) + except OSError as exc: + result.status = RunStatus.ERROR + result.message = _describe_os_error(exc) + else: + if result.failed: + result.status = RunStatus.PARTIAL + return result + + def _execute( + self, + rule: RuleSnapshot, + src_root: Path, + dest: Destination, + token: CancelToken, + result: RunResult, + dry_run: bool, + quarantine_root: Path | None, + ) -> None: + if not src_root.is_dir(): + raise RunAborted(f"Quellordner {src_root} existiert nicht") + if not dry_run: + dest.prepare() + token.check() + + exclude: set[Path] = set() + if quarantine_root is not None: + exclude.add(quarantine_root.resolve()) + if isinstance(dest, LocalDestination): + exclude.add(dest.root.resolve()) # Ziel innerhalb der Quelle nie erneut scannen + matcher = Matcher(rule.pattern, rule.pattern_type) + candidates = list(scan(src_root, matcher, rule.recursive, frozenset(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. + done = self.ledger.copied_keys(rule.id) + if not dry_run: + self.ledger.prune_copied(rule.id, {c.relpath for c in candidates}) + fresh = [c for c in candidates if (c.relpath, c.state.size, c.state.mtime_ns) not in done] + unchanged = len(candidates) - len(fresh) + + pending: list[Candidate] = [] + now = self._clock() + for c in fresh: + if rule.max_size_mb is not None and c.state.size > rule.max_size_mb * 1024 * 1024: + result.add( + ItemResult( + c.relpath, + ItemStatus.SKIPPED, + size=c.state.size, + message=f"größer als {rule.max_size_mb} MB", + ) + ) + elif rule.min_age_seconds and now - c.state.mtime_ns / 1e9 < rule.min_age_seconds: + result.add( + ItemResult( + c.relpath, + ItemStatus.SKIPPED, + size=c.state.size, + message=f"jünger als {rule.min_age_seconds} s", + ) + ) + else: + pending.append(c) + + # Stabilität (Fallstrick 1): zweite Messung nach dem Stabilitätsfenster. + if pending and rule.stability_seconds > 0: + self._sleep(rule.stability_seconds) + token.check() + stable: list[Candidate] = [] + for c in pending: + st = current_state(c.path) + if st is None: + continue # inzwischen verschwunden – nichts zu tun + if st != c.state: + result.add( + ItemResult( + c.relpath, + ItemStatus.SKIPPED, + size=st.size, + message="wird noch geschrieben (Größe/Zeit geändert)", + ) + ) + continue + stable.append(c) + + failures = self.ledger.failures(rule.id) + for c in stable: + token.check() + result.add(self._process(rule, c, dest, token, dry_run, quarantine_root, failures)) + + if unchanged: + result.message = f"{unchanged} bereits übertragen" + + # -- einzelne Datei ------------------------------------------------------------------ + + def _meta( + self, rule: RuleSnapshot, path: Path, dest: Destination + ) -> tuple[MetaOptions, list[str]]: + st = os.stat(path) + notes: list[str] = [] + owner = rule.set_owner + if owner is not None: + if dest.kind == "cifs": + owner = None + if self._warn_once(f"{rule.id}:cifs-chown"): + notes.append( + "CIFS-Ziel: Eigentümer bestimmen die Mount-Optionen, chown übersprungen" + ) + elif dest.kind == "nfs" and rule.dst_root_squash is not False: + owner = None + if self._warn_once(f"{rule.id}:nfs-chown"): + reason = "Root-Squash aktiv" if rule.dst_root_squash else "Schreibtest fehlt" + notes.append(f"NFS-Ziel: chown übersprungen ({reason})") + mode = rule.set_mode if rule.set_mode is not None else stat.S_IMODE(st.st_mode) + return MetaOptions(st.st_atime_ns, st.st_mtime_ns, mode, owner), notes + + def _process( + self, + rule: RuleSnapshot, + c: Candidate, + dest: Destination, + token: CancelToken, + dry_run: bool, + quarantine_root: Path | None, + failures: dict[str, FailureInfo], + ) -> ItemResult: + size = c.state.size + failure = failures.get(c.relpath) + + # Quarantäne für wiederholt scheiternde Dateien + if ( + quarantine_root is not None + and failure is not None + and failure.attempts >= rule.quarantine_after + and (failure.size, failure.mtime_ns) == (c.state.size, c.state.mtime_ns) + ): + if dry_run: + return ItemResult( + c.relpath, + ItemStatus.PLANNED, + str(quarantine_root / c.relpath), + size, + "würde in Quarantäne verschoben", + ) + return self._quarantine(rule, c, quarantine_root, 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" + return ItemResult( + c.relpath, ItemStatus.PLANNED, dest.describe(planned), size, msg or f"würde {verb}" + ) + + 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))})", + ) + + def verify() -> None: + if current_state(c.path) != c.state: + raise SourceChanged("Quelle hat sich während der Übertragung geändert") + + try: + meta, notes = self._meta(rule, c.path, dest) + res = None + if rule.action is RuleAction.MOVE: + res = dest.move_same_device(c.path, c.relpath, rule.on_collision, meta) + if res is None: + res = dest.commit(c.path, size, c.relpath, rule.on_collision, meta, verify, token) + 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) + + notes += res.notes + target = dest.describe(res.final_rel) + if c.relpath in failures: + self.ledger.clear_failure(rule.id, c.relpath) + + 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: + try: + os.unlink(c.path) + 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) + return ItemResult( + c.relpath, + ItemStatus.FAILED, + target, + size, + f"Ziel geschrieben, aber Quelle nicht löschbar: {_describe_os_error(exc)} – " + "Datei wird nicht erneut übertragen", + ) + return ItemResult(c.relpath, ItemStatus.MOVED, target, size, "; ".join(notes) or None) + + def _fail(self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool) -> ItemResult: + if isinstance(exc, OSError) and exc.errno in MOUNT_ERRNOS: + raise RunAborted(f"Anbindung gestört: {_describe_os_error(exc)}") from exc + msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc) + if not dry_run: + attempts = self.ledger.record_failure( + rule.id, c.relpath, c.state.size, c.state.mtime_ns, msg + ) + msg = f"{msg} (Versuch {attempts})" + 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 _quarantine( + self, rule: RuleSnapshot, c: Candidate, root: Path, attempts: int + ) -> ItemResult: + try: + for cand in collision_candidates(c.relpath): + target = root / cand + if os.path.lexists(target): + continue + target.parent.mkdir(mode=0o755, parents=True, exist_ok=True) + os.rename(c.path, target) + break + else: + raise TransferError("kein freier Name im Quarantäneordner") + except (OSError, TransferError) as exc: + msg = _describe_os_error(exc) if isinstance(exc, OSError) else str(exc) + return ItemResult( + c.relpath, ItemStatus.FAILED, None, c.state.size, f"Quarantäne gescheitert: {msg}" + ) + self.ledger.clear_failure(rule.id, c.relpath) + event(log, "file.quarantined", level=30, rule=rule.name, file=c.relpath) + return ItemResult( + c.relpath, + ItemStatus.QUARANTINED, + str(target), + c.state.size, + f"nach {attempts} Fehlversuchen in Quarantäne verschoben", + ) + + +__all__ = ["Engine", "FileState", "ItemResult", "RuleSnapshot", "RunAborted", "RunResult"] diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..e338f30 --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,680 @@ +"""Mover-Engine mit lokalen Zielen – Fallstricke 1, 3, 4, 10, 11, 14 und Quarantäne.""" + +from __future__ import annotations + +import errno +import os +import time +from collections.abc import Callable +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 ( + CollisionPolicy, + CopiedFile, + ItemStatus, + PatternType, + Rule, + RuleAction, + RunStatus, +) +from netmover.mover import destinations as dest_mod +from netmover.mover.destinations import LocalDestination, TransferError +from netmover.mover.engine import Engine, RuleSnapshot, RunResult +from netmover.mover.ledger import Ledger +from netmover.mover.naming import collision_candidates +from netmover.mover.scan import Matcher +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 + + +@pytest.fixture +def ledger(session_factory: sessionmaker[Session]) -> Ledger: + return Ledger(session_factory) + + +Sleeper = Callable[[float], None] + + +@pytest.fixture +def hooks() -> list[Callable[[], None]]: + """Aktionen, die während des Stabilitätsfensters ausgeführt werden.""" + return [] + + +@pytest.fixture +def engine(ledger: Ledger, hooks: list[Callable[[], None]]) -> Engine: + def sleep(_: float) -> None: + for h in hooks: + h() + + return Engine(ledger, sleep=sleep) + + +@pytest.fixture +def cross_device(monkeypatch: pytest.MonkeyPatch) -> None: + """tmp_path liegt auf einem Gerät – für Kopier-Tests den Schnellweg abschalten.""" + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + + +@pytest.fixture +def src(tmp_path: Path) -> Path: + p = tmp_path / "src" + p.mkdir() + return p + + +@pytest.fixture +def dst(tmp_path: Path) -> Path: + p = tmp_path / "dst" + p.mkdir() + return p + + +def snap(rule_id: int, **kw: object) -> RuleSnapshot: + return RuleSnapshot(id=rule_id, name="r", **kw) # type: ignore[arg-type] + + +def run( + engine: Engine, rule: RuleSnapshot, src: Path, dst: Path | LocalDestination, **kw: object +) -> RunResult: + d = dst if isinstance(dst, LocalDestination) else LocalDestination(dst) + return engine.execute(rule, src, d, CancelToken(), **kw) # type: ignore[arg-type] + + +def write(path: Path, data: str = "inhalt") -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(data) + return path + + +def statuses(result: RunResult) -> dict[str, ItemStatus]: + return {i.src: i.status for i in result.items} + + +# -- Fallstrick 1: halb geschriebene Dateien -------------------------------------------- + + +def test_growing_file_is_not_touched( + engine: Engine, rule_id: int, src: Path, dst: Path, hooks: list[Callable[[], None]] +) -> None: + write(src / "fertig.pdf", "a" * 100) + growing = write(src / "wachsend.pdf", "b" * 10) + hooks.append(lambda: growing.write_text("b" * 20)) # wächst während des Fensters + result = run(engine, snap(rule_id), src, dst) + assert statuses(result) == {"fertig.pdf": ItemStatus.MOVED, "wachsend.pdf": ItemStatus.SKIPPED} + assert growing.read_text() == "b" * 20 + assert not (dst / "wachsend.pdf").exists() + assert (dst / "fertig.pdf").read_text() == "a" * 100 + + +def test_mtime_change_alone_counts_as_unstable( + engine: Engine, rule_id: int, src: Path, dst: Path, hooks: list[Callable[[], None]] +) -> None: + f = write(src / "scan.pdf") + st = f.stat() + hooks.append(lambda: os.utime(f, ns=(st.st_atime_ns, st.st_mtime_ns + 1_000_000))) + result = run(engine, snap(rule_id), src, dst) + assert statuses(result) == {"scan.pdf": ItemStatus.SKIPPED} + assert f.exists() + + +def test_stability_window_is_waited(ledger: Ledger, rule_id: int, src: Path, dst: Path) -> None: + waited: list[float] = [] + eng = Engine(ledger, sleep=waited.append) + write(src / "a.pdf") + run(eng, snap(rule_id, stability_seconds=7), src, dst) + assert waited == [7] + waited.clear() + run(eng, snap(rule_id, stability_seconds=7), src, dst) # nichts mehr da → kein Warten + assert waited == [] + + +@pytest.mark.parametrize( + "name", + [".versteckt.pdf", "scan.pdf.part", "x.tmp", "dl.crdownload", "f.filepart", "t.!ut", "A.PART"], +) +def test_hidden_and_partial_files_are_ignored( + engine: Engine, rule_id: int, src: Path, dst: Path, name: str +) -> None: + write(src / name) + result = run(engine, snap(rule_id), src, dst) + assert result.seen == 0 + assert (src / name).exists() + + +def test_min_age_and_max_size(engine: Engine, rule_id: int, src: Path, dst: Path) -> None: + write(src / "neu.pdf") + old = write(src / "alt.pdf") + os.utime(old, (time.time() - 3600, time.time() - 3600)) + write(src / "gross.pdf", "x" * (2 * 1024 * 1024)) + os.utime(src / "gross.pdf", (time.time() - 3600, time.time() - 3600)) + result = run(engine, snap(rule_id, min_age_seconds=600, max_size_mb=1), src, dst) + assert statuses(result) == { + "alt.pdf": ItemStatus.MOVED, + "gross.pdf": ItemStatus.SKIPPED, + "neu.pdf": ItemStatus.SKIPPED, + } + + +# -- Fallstrick 3: Atomarität --------------------------------------------------------------- + + +@pytest.mark.usefixtures("cross_device") +def test_final_name_never_shows_partial_content( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + data = "x" * (3 * dest_mod.CHUNK + 17) + write(src / "gross.pdf", data) + seen_during_write: list[list[str]] = [] + original = LocalDestination._upload + + def spy(self: LocalDestination, *a: object, **kw: object) -> list[str]: + notes = original(self, *a, **kw) # type: ignore[arg-type] + seen_during_write.append(sorted(p.name for p in dst.iterdir())) + return notes + + monkeypatch.setattr(LocalDestination, "_upload", spy) + result = run(engine, snap(rule_id), src, dst) + assert statuses(result) == {"gross.pdf": ItemStatus.MOVED} + # während/nach dem Schreiben existiert nur die .part-Datei, nicht der Endname + (names,) = seen_during_write + assert len(names) == 1 and names[0].startswith(".netmover-") and names[0].endswith(".part") + assert (dst / "gross.pdf").read_text() == data + assert not (src / "gross.pdf").exists() + + +STAGES = ["_upload", "_apply_meta", "_link_noreplace", "verify"] + + +@pytest.mark.usefixtures("cross_device") +@pytest.mark.parametrize("stage", STAGES) +def test_abort_at_any_stage_keeps_source_and_leaves_no_part( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch, stage: str +) -> None: + write(src / "scan.pdf", "wichtig") + if stage == "verify": + # Quelle wird zwischen Kopie und Umbenennen verändert + original = LocalDestination._apply_meta + + def touch(self: LocalDestination, rel: str, meta: object) -> list[str]: + (src / "scan.pdf").write_text("wichtig!!") + return original(self, rel, meta) # type: ignore[arg-type] + + monkeypatch.setattr(LocalDestination, "_apply_meta", touch) + else: + + def boom(self: object, *a: object, **kw: object) -> None: + raise OSError(errno.EIO if stage == "_upload" else errno.ENOSPC, "Fehler") + + monkeypatch.setattr(LocalDestination, stage, boom) + result = run(engine, snap(rule_id), src, dst) + assert (src / "scan.pdf").exists() + assert list(dst.iterdir()) == [] + item = result.items[0] if result.items else None + if stage == "_upload": # EIO → Anbindung gestört, ganzer Lauf abgebrochen + assert result.status is RunStatus.ERROR + elif stage == "verify": + assert item is not None and item.status is ItemStatus.SKIPPED + else: + assert item is not None and item.status is ItemStatus.FAILED + + +@pytest.mark.parametrize("path", ["same-device", "copy"]) +def test_source_not_deletable_keeps_target_and_prevents_duplicate( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch, path: str +) -> None: + if path == "copy": + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + write(src / "scan.pdf") + real_unlink = os.unlink + + def deny(path: object, *a: object, **kw: object) -> None: + if str(path).startswith(str(src)): + raise PermissionError(errno.EACCES, "Permission denied", str(path)) + real_unlink(path) # type: ignore[arg-type] + + monkeypatch.setattr("netmover.mover.engine.os.unlink", deny) + first = run(engine, snap(rule_id), src, dst) + assert first.items[0].status is ItemStatus.FAILED + assert "nicht löschbar" in (first.items[0].message or "") + assert (dst / "scan.pdf").exists() and (src / "scan.pdf").exists() + second = run(engine, snap(rule_id), src, dst) + assert second.items == [] and second.message == "1 bereits übertragen" + assert sorted(p.name for p in dst.iterdir()) == ["scan.pdf"] + + +def test_stale_part_files_are_cleaned(engine: Engine, rule_id: int, src: Path, dst: Path) -> None: + old = write(dst / ".netmover-dead.part", "rest") + young = write(dst / ".netmover-busy.part", "rest") + os.utime(old, (time.time() - 7200, time.time() - 7200)) + run(engine, snap(rule_id), src, dst) + assert not old.exists() + assert young.exists() + + +def test_same_device_move_uses_link_without_copy( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + write(src / "a.pdf", "x") + ino = (src / "a.pdf").stat().st_ino + + def no_copy(*a: object, **kw: object) -> None: + raise AssertionError("darf auf demselben Gerät nicht kopieren") + + monkeypatch.setattr(LocalDestination, "_upload", no_copy) + result = run(engine, snap(rule_id), src, dst) + assert statuses(result) == {"a.pdf": ItemStatus.MOVED} + assert (dst / "a.pdf").stat().st_ino == ino + assert not (src / "a.pdf").exists() + + +def test_cross_device_path_is_used_when_devices_differ( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + write(src / "a.pdf", "x") + ino = (src / "a.pdf").stat().st_ino + result = run(engine, snap(rule_id), src, dst) + assert statuses(result) == {"a.pdf": ItemStatus.MOVED} + assert (dst / "a.pdf").stat().st_ino != ino + + +def test_filesystem_without_hardlinks_falls_back_to_rename( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + + def no_link(*a: object, **kw: object) -> None: + raise OSError(errno.EPERM, "Operation not permitted") + + monkeypatch.setattr("netmover.mover.destinations.os.link", no_link) + write(src / "a.pdf") + write(dst / "a.pdf", "alt") + result = run(engine, snap(rule_id), src, dst) + assert result.items[0].dst == str(dst / "a-1.pdf") + assert (dst / "a.pdf").read_text() == "alt" + + +# -- Fallstrick 4: Namenskollisionen --------------------------------------------------------- + + +def test_collision_candidates() -> None: + gen = collision_candidates("sub/scan001.pdf") + assert [next(gen) for _ in range(3)] == [ + "sub/scan001.pdf", + "sub/scan001-1.pdf", + "sub/scan001-2.pdf", + ] + gen = collision_candidates("README") + assert [next(gen) for _ in range(2)] == ["README", "README-1"] + + +def test_rename_counts_up_against_filesystem( + engine: Engine, rule_id: int, src: Path, dst: Path +) -> None: + write(dst / "scan001.pdf", "alt") + for i in range(3): + write(src / "scan001.pdf", f"neu{i}") + run(engine, snap(rule_id), src, dst) + assert sorted(p.name for p in dst.iterdir()) == [ + "scan001-1.pdf", + "scan001-2.pdf", + "scan001-3.pdf", + "scan001.pdf", + ] + assert (dst / "scan001.pdf").read_text() == "alt" + assert (dst / "scan001-3.pdf").read_text() == "neu2" + + +def test_rename_rechecks_filesystem_each_time( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Zwischen Plan und Umbenennen belegt ein Dritter den Namen – kein Überschreiben.""" + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + write(dst / "scan.pdf", "alt") + write(src / "scan.pdf", "neu") + original = LocalDestination._apply_meta + + def intruder(self: LocalDestination, rel: str, meta: object) -> list[str]: + write(dst / "scan-1.pdf", "fremd") + return original(self, rel, meta) # type: ignore[arg-type] + + monkeypatch.setattr(LocalDestination, "_apply_meta", intruder) + result = run(engine, snap(rule_id), src, dst) + assert result.items[0].dst == str(dst / "scan-2.pdf") + assert (dst / "scan-1.pdf").read_text() == "fremd" + assert (dst / "scan-2.pdf").read_text() == "neu" + + +def test_skip_policy(engine: Engine, rule_id: int, src: Path, dst: Path) -> None: + write(dst / "a.pdf", "alt") + write(src / "a.pdf", "neu") + result = run(engine, snap(rule_id, on_collision=CollisionPolicy.SKIP), src, dst) + assert statuses(result) == {"a.pdf": ItemStatus.SKIPPED} + assert (dst / "a.pdf").read_text() == "alt" + assert (src / "a.pdf").exists() + + +def test_overwrite_policy( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + write(dst / "a.pdf", "alt") + write(src / "a.pdf", "neu") + result = run(engine, snap(rule_id, on_collision=CollisionPolicy.OVERWRITE), src, dst) + assert statuses(result) == {"a.pdf": ItemStatus.MOVED} + assert (dst / "a.pdf").read_text() == "neu" + + +# -- Fallstrick 11: Kopiermodus mit Gedächtnis --------------------------------------------- + + +def test_copy_mode_copies_each_version_once( + engine: Engine, rule_id: int, src: Path, dst: Path, session_factory: sessionmaker[Session] +) -> None: + rule = snap(rule_id, action=RuleAction.COPY) + f = write(src / "scan.pdf", "v1") + for _ in range(3): + run(engine, rule, src, dst) + assert sorted(p.name for p in dst.iterdir()) == ["scan.pdf"] + assert f.exists() + + # neue Version → läuft durch die Kollisionsregel + f.write_text("version 2") + os.utime(f, ns=(f.stat().st_atime_ns, f.stat().st_mtime_ns + 5_000_000_000)) + result = run(engine, rule, src, dst) + assert statuses(result) == {"scan.pdf": ItemStatus.COPIED} + run(engine, rule, src, dst) + assert sorted(p.name for p in dst.iterdir()) == ["scan-1.pdf", "scan.pdf"] + assert (dst / "scan-1.pdf").read_text() == "version 2" + + # Quelle verschwindet → Buchführung wird bereinigt + f.unlink() + run(engine, rule, src, dst) + with session_factory() as s: + assert s.scalars(select(CopiedFile)).all() == [] + + +def test_copy_mode_preserves_mtime(engine: Engine, rule_id: int, src: Path, dst: Path) -> None: + f = write(src / "a.pdf") + os.utime(f, ns=(1_600_000_000_000_000_000, 1_600_000_000_123_456_789)) + run(engine, snap(rule_id, action=RuleAction.COPY), src, dst) + assert (dst / "a.pdf").stat().st_mtime_ns == 1_600_000_000_123_456_789 + + +# -- Fallstrick 14: Platz --------------------------------------------------------------------- + + +def test_insufficient_space_skips_file_without_partial_copy( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + write(src / "a.pdf", "x" * 1000) + monkeypatch.setattr(LocalDestination, "free_bytes", lambda self: 1040) # < 1000 * 1.05 + result = run(engine, snap(rule_id), src, dst) + assert result.items[0].status is ItemStatus.FAILED + assert "zu wenig Platz" in (result.items[0].message or "") + assert list(dst.iterdir()) == [] + assert (src / "a.pdf").exists() + + +def test_metadata_failure_is_only_a_hint( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + + def deny(*a: object, **kw: object) -> None: + raise PermissionError(errno.EPERM, "Operation not permitted") + + monkeypatch.setattr("netmover.mover.destinations.os.utime", deny) + write(src / "a.pdf") + result = run(engine, snap(rule_id), src, dst) + assert result.items[0].status is ItemStatus.MOVED + assert "Änderungszeit nicht übernommen" in (result.items[0].message or "") + + +# -- Fallstrick 10: Berechtigungen --------------------------------------------------------- + + +def test_set_mode_applied(engine: Engine, rule_id: int, src: Path, dst: Path) -> None: + write(src / "a.pdf").chmod(0o600) + run(engine, snap(rule_id, set_mode=0o640), src, dst) + assert (dst / "a.pdf").stat().st_mode & 0o777 == 0o640 + + +def test_unchanged_mode_is_copied_from_source( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + write(src / "a.pdf").chmod(0o604) + run(engine, snap(rule_id), src, dst) + assert (dst / "a.pdf").stat().st_mode & 0o777 == 0o604 + + +def test_chown_without_permission_is_clear_message( + engine: Engine, rule_id: int, src: Path, dst: Path +) -> None: + if os.geteuid() == 0: + pytest.skip("als root ist chown erlaubt") + write(src / "a.pdf") + result = run(engine, snap(rule_id, set_owner=(0, 0)), src, dst) + assert result.items[0].status is ItemStatus.MOVED + assert "keine Berechtigung" in (result.items[0].message or "") + + +@pytest.mark.parametrize(("kind", "squash"), [("cifs", None), ("nfs", True), ("nfs", None)]) +def test_chown_skipped_on_cifs_and_squashed_nfs_warns_once( + engine: Engine, + rule_id: int, + src: Path, + dst: Path, + monkeypatch: pytest.MonkeyPatch, + kind: str, + squash: bool | None, +) -> None: + called: list[object] = [] + monkeypatch.setattr("netmover.mover.destinations.os.chown", lambda *a: called.append(a)) + write(src / "a.pdf") + write(src / "b.pdf") + d = LocalDestination(dst, kind) # type: ignore[arg-type] + result = run(engine, snap(rule_id, set_owner=(1000, 1000), dst_root_squash=squash), src, d) + assert called == [] + msgs = [i.message for i in result.items if i.message and "chown übersprungen" in i.message] + assert len(msgs) == 1 # einmalig warnen + + +def test_chown_on_nfs_without_squash_is_tried( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + called: list[object] = [] + monkeypatch.setattr("netmover.mover.destinations.os.chown", lambda *a: called.append(a)) + write(src / "a.pdf") + run( + engine, + snap(rule_id, set_owner=(1000, 100), dst_root_squash=False), + src, + LocalDestination(dst, "nfs"), + ) + assert called and called[0][1:] == (1000, 100) # type: ignore[index] + + +# -- Filter, Rekursion, Probelauf, Quarantäne --------------------------------------------- + + +def test_matcher() -> None: + m = Matcher("*.pdf; *.JPG", PatternType.GLOB) + assert m("a/Scan.PDF") and m("b.jpg") and not m("c.txt") + r = Matcher(r"^rechnungen/.*\.pdf$", PatternType.REGEX) + assert r("rechnungen/2026.pdf") and not r("scan/2026.pdf") + + +def test_recursive_preserves_structure_and_skips_synology_dirs( + engine: Engine, rule_id: int, src: Path, dst: Path +) -> None: + write(src / "max" / "a.pdf") + write(src / "@eaDir" / "a.pdf" / "SYNOFILE_THUMB_M.jpg") + write(src / "#recycle" / "alt.pdf") + write(src / "top.pdf") + flat = run(engine, snap(rule_id), src, dst) + assert statuses(flat) == {"top.pdf": ItemStatus.MOVED} + rec = run(engine, snap(rule_id, recursive=True), src, dst) + assert statuses(rec) == {"max/a.pdf": ItemStatus.MOVED} + assert (dst / "max" / "a.pdf").exists() + assert (src / "#recycle" / "alt.pdf").exists() + + +def test_symlinks_are_ignored( + engine: Engine, rule_id: int, src: Path, dst: Path, tmp_path: Path +) -> None: + secret = write(tmp_path / "geheim.txt") + (src / "link.txt").symlink_to(secret) + result = run(engine, snap(rule_id), src, dst) + assert result.seen == 0 and secret.exists() + + +def test_dry_run_touches_nothing( + engine: Engine, rule_id: int, src: Path, dst: Path, session_factory: sessionmaker[Session] +) -> None: + write(src / "a.pdf") + write(dst / "a.pdf", "alt") + write(src / "b.pdf") + result = run(engine, snap(rule_id, action=RuleAction.COPY), src, dst, dry_run=True) + assert {i.src: (i.status, i.dst) for i in result.items} == { + "a.pdf": (ItemStatus.PLANNED, str(dst / "a-1.pdf")), + "b.pdf": (ItemStatus.PLANNED, str(dst / "b.pdf")), + } + assert sorted(p.name for p in dst.iterdir()) == ["a.pdf"] + assert sorted(p.name for p in src.iterdir()) == ["a.pdf", "b.pdf"] + with session_factory() as s: + assert s.scalars(select(CopiedFile)).all() == [] + + +def test_quarantine_after_repeated_failures( + engine: Engine, + rule_id: int, + src: Path, + dst: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + + def broken(self: object, *a: object, **kw: object) -> None: + raise TransferError("kaputt") + + monkeypatch.setattr(LocalDestination, "_link_noreplace", broken) + write(src / "defekt.pdf") + quarantine = src / "quarantaene" + rule = snap(rule_id, quarantine_after=3, recursive=True) + msgs = [] + for _ in range(3): + r = run(engine, rule, src, dst, quarantine_root=quarantine) + msgs.append(r.items[0].message) + assert msgs[-1] == "kaputt (Versuch 3)" + r = run(engine, rule, src, dst, quarantine_root=quarantine) + assert statuses(r) == {"defekt.pdf": ItemStatus.QUARANTINED} + assert (quarantine / "defekt.pdf").exists() + # Quarantäne liegt in der Quelle, wird aber trotz Rekursion nicht erneut gescannt + r = run(engine, rule, src, dst, quarantine_root=quarantine) + assert r.seen == 0 + + +def test_mount_error_mid_run_aborts_instead_of_failing_every_file( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None) + + def stale(self: object, *a: object, **kw: object) -> None: + raise OSError(errno.ESTALE, "Stale file handle") + + monkeypatch.setattr(LocalDestination, "_upload", stale) + write(src / "a.pdf") + write(src / "b.pdf") + result = run(engine, snap(rule_id), src, dst) + assert result.status is RunStatus.ERROR + assert "gestört" in (result.message or "") + assert (src / "a.pdf").exists() and (src / "b.pdf").exists() + + +def test_cancel_stops_before_next_file(engine: Engine, rule_id: int, src: Path, dst: Path) -> None: + write(src / "a.pdf") + token = CancelToken() + token.cancel() + result = engine.execute(snap(rule_id), src, LocalDestination(dst), token) + assert result.status is RunStatus.ERROR + assert (src / "a.pdf").exists() + + +def test_missing_source_dir(engine: Engine, rule_id: int, tmp_path: Path, dst: Path) -> None: + result = run(engine, snap(rule_id), tmp_path / "fehlt", dst) + assert result.status is RunStatus.ERROR and "existiert nicht" in (result.message or "") + + +def test_local_destination_must_exist( + engine: Engine, rule_id: int, src: Path, tmp_path: Path +) -> None: + write(src / "a.pdf") + result = run(engine, snap(rule_id), src, LocalDestination(tmp_path / "fehlt")) + assert result.status is RunStatus.ERROR + assert (src / "a.pdf").exists() + + +@pytest.mark.usefixtures("cross_device") +def test_fsync_before_final_rename( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + events: list[str] = [] + real_fsync, real_link = os.fsync, os.link + + def fsync(fd: int) -> None: + events.append("fsync") + real_fsync(fd) + + def link(a: object, b: object, *args: object, **kw: object) -> None: + events.append(f"link:{Path(str(b)).name}") + real_link(a, b) # type: ignore[arg-type] + + monkeypatch.setattr("netmover.mover.destinations.os.fsync", fsync) + monkeypatch.setattr("netmover.mover.destinations.os.link", link) + write(src / "a.pdf") + run(engine, snap(rule_id), src, dst) + assert events.index("fsync") < events.index("link:a.pdf") + + +@pytest.mark.usefixtures("cross_device") +def test_name_taken_between_check_and_link_is_never_overwritten( + engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Enges Wettlauf-Fenster: exists() sagt „frei“, link() meldet EEXIST.""" + write(dst / "scan.pdf", "fremd") + write(src / "scan.pdf", "neu") + real_exists = LocalDestination.exists + lied: list[str] = [] + + def stale_exists(self: LocalDestination, rel: str) -> bool: + if rel == "scan.pdf" and not lied and self.root == dst: + lied.append(rel) + return False # veraltete Antwort + return real_exists(self, rel) + + monkeypatch.setattr(LocalDestination, "plan", lambda self, rel, policy: rel) + monkeypatch.setattr(LocalDestination, "exists", stale_exists) + result = run(engine, snap(rule_id), src, dst) + assert lied + assert result.items[0].dst == str(dst / "scan-1.pdf") + assert (dst / "scan.pdf").read_text() == "fremd" + assert (dst / "scan-1.pdf").read_text() == "neu"