From e59932ab7502993a99b2715bc0d0e553e9352935 Mon Sep 17 00:00:00 2001 From: Florian Wilde Date: Mon, 12 Oct 2026 00:22:39 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Archiv=20je=20Regel=20=E2=80=93=20erst?= =?UTF-8?q?=20Archiv,=20dann=20Ziel,=20dann=20Quelle=20l=C3=B6schen?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Zusätzliche Kopie ins Archiv (Anbindung oder lokaler Pfad, Ablage wie Quelle oder nach Datum). Buchführung verhindert doppelte Archivkopien bei Neuversuch; im Archiv wird nie überschrieben. Migration 0004 mit Tests auf befüllten Datenbanken (inkl. Downgrade). Co-Authored-By: Claude Opus 5.5 --- .../migrations/versions/0004_archive.py | 103 +++++ src/netmover/models.py | 41 ++ src/netmover/mover/engine.py | 94 ++++- src/netmover/mover/ledger.py | 49 ++- src/netmover/mover/runner.py | 97 +++-- tests/test_archive.py | 351 ++++++++++++++++++ 6 files changed, 701 insertions(+), 34 deletions(-) create mode 100644 src/netmover/migrations/versions/0004_archive.py create mode 100644 tests/test_archive.py diff --git a/src/netmover/migrations/versions/0004_archive.py b/src/netmover/migrations/versions/0004_archive.py new file mode 100644 index 0000000..3e244f0 --- /dev/null +++ b/src/netmover/migrations/versions/0004_archive.py @@ -0,0 +1,103 @@ +"""archive + +Revision ID: 0004 +Revises: 0003 +Create Date: 2026-10-12 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +import netmover.models + +revision: str = "0004" +down_revision: str | None = "0003" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + 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"), + ), + ) + with op.batch_alter_table("rules", schema=None) as batch_op: + batch_op.add_column( + sa.Column( + "archive_kind", + sa.Enum( + "none", + "mount", + "local", + name="archive_kind", + native_enum=False, + create_constraint=True, + length=24, + ), + 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", + sa.Enum( + "flat", + "date", + name="archive_layout", + native_enum=False, + create_constraint=True, + length=24, + ), + 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", + ) + + +def downgrade() -> None: + 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") + # CHECK-Regeln der Auswahlfelder zuerst entfernen – sonst verweisen sie beim + # Neuaufbau der Tabelle auf Spalten, die es nicht mehr gibt. + 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") + + op.drop_table("archived_files") diff --git a/src/netmover/models.py b/src/netmover/models.py index 2d6d96d..89c7fd2 100644 --- a/src/netmover/models.py +++ b/src/netmover/models.py @@ -10,6 +10,7 @@ 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. * ``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. @@ -138,6 +139,17 @@ class CollisionPolicy(enum.StrEnum): OVERWRITE = "overwrite" +class ArchiveKind(enum.StrEnum): + NONE = "none" + MOUNT = "mount" + LOCAL = "local" + + +class ArchiveLayout(enum.StrEnum): + FLAT = "flat" # wie in der Quelle + DATE = "date" # Jahr/Monat/…, nach Tag der Verarbeitung + + class RunStatus(enum.StrEnum): RUNNING = "running" OK = "ok" # keine Dateifehler @@ -259,6 +271,18 @@ 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) @@ -269,6 +293,7 @@ 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]) runs: Mapped[list[Run]] = relationship( back_populates="rule", cascade="all, delete-orphan", passive_deletes=True ) @@ -330,6 +355,22 @@ 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.""" + + __tablename__ = "archived_files" + __table_args__ = (UniqueConstraint("rule_id", "src_relpath", "size", "mtime_ns"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + rule_id: Mapped[int] = mapped_column(ForeignKey("rules.id", ondelete="CASCADE")) + 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) + + class FileFailure(Base): """Fehlversuche pro Datei, Grundlage für die Quarantäne.""" diff --git a/src/netmover/mover/engine.py b/src/netmover/mover/engine.py index 37b995a..41b4e58 100644 --- a/src/netmover/mover/engine.py +++ b/src/netmover/mover/engine.py @@ -19,6 +19,7 @@ import posixpath import time from collections.abc import Callable from dataclasses import dataclass, field +from datetime import datetime from pathlib import Path from typing import Any @@ -72,6 +73,7 @@ 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 @@ -159,6 +161,7 @@ class Engine: dry_run: bool = False, quarantine_rel: str | None = None, quarantine_root: Path | None = None, + archive: Destination | None = None, ) -> RunResult: """``source``: eine :class:`Source` oder (Kurzform) ein lokales Verzeichnis. Quarantäne relativ zur Wurzel der Anbindung (``quarantine_rel``) oder – bei @@ -168,7 +171,7 @@ class Engine: 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) + self._execute(rule, src, dest, token, result, dry_run, quarantine_rel, archive) except Cancelled: result.status = RunStatus.ERROR result.message = "Lauf abgebrochen" @@ -195,10 +198,13 @@ class Engine: result: RunResult, dry_run: bool, quarantine_rel: str | None, + archive: Destination | None = None, ) -> None: src.check_root() if not dry_run: dest.prepare() + if archive is not None: + archive.prepare() token.check() exclude = frozenset({quarantine_rel}) if quarantine_rel else frozenset() @@ -212,8 +218,12 @@ class Engine: # 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) + archived = self.ledger.archived_keys(rule.id) if archive is not None else set() if not dry_run: - self.ledger.prune_copied(rule.id, {c.relpath for c in candidates}) + 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] unchanged = len(candidates) - len(fresh) @@ -271,7 +281,11 @@ 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)) + result.add( + self._process( + rule, c, src, dest, token, dry_run, quarantine_rel, failures, archive, archived + ) + ) if unchanged: result.message = f"{unchanged} bereits übertragen" @@ -309,8 +323,14 @@ class Engine: dry_run: bool, quarantine_rel: str | None, failures: dict[str, FailureInfo], + archive: Destination | None = None, + archived: set[tuple[str, int, int]] | None = None, ) -> ItemResult: 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()) + ) failure = failures.get(c.relpath) # Quarantäne für wiederholt scheiternde Dateien @@ -349,9 +369,14 @@ class Engine: 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}" - ) + 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() @@ -367,10 +392,51 @@ class Engine: f"benötigt {_mb(int(size * SPACE_RESERVE))})", ) + if need_archive and archive is not None: + try: + afree = archive.free_bytes() + except OSError: + afree = None + if afree is not None and afree < size * SPACE_RESERVE: + return ItemResult( + c.relpath, + ItemStatus.FAILED, + archive.describe(arch_rel), + size, + f"zu wenig Platz im Archiv (frei {_mb(afree)}, " + f"benötigt {_mb(int(size * SPACE_RESERVE))})", + ) + 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: + try: + meta, _ = self._meta(rule, src, c.relpath, dest) + ares = archive.commit( + _SourceFile(src, c.relpath), + size, + arch_rel, + CollisionPolicy.RENAME, + 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") + 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}" + try: meta, notes = self._meta(rule, src, c.relpath, dest) res = None @@ -401,6 +467,8 @@ class Engine: return self._fail(rule, c, exc, dry_run) 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) @@ -425,10 +493,12 @@ class Engine: ) 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: + def _fail( + self, rule: RuleSnapshot, c: Candidate, exc: Exception, dry_run: bool, prefix: str = "" + ) -> 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) + raise RunAborted(f"{prefix}Anbindung gestört: {_describe_os_error(exc)}") from exc + msg = prefix + (_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 @@ -437,6 +507,12 @@ 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: + if layout == "date": + t = datetime.fromtimestamp(self._clock()) + return f"{t:%Y}/{t:%m}/{relpath}" + return relpath + def _quarantine( self, rule: RuleSnapshot, c: Candidate, src: Source, q_rel: str, attempts: int ) -> ItemResult: diff --git a/src/netmover/mover/ledger.py b/src/netmover/mover/ledger.py index 09c3767..424fdf1 100644 --- a/src/netmover/mover/ledger.py +++ b/src/netmover/mover/ledger.py @@ -14,7 +14,7 @@ from sqlalchemy import delete, select from sqlalchemy.orm import Session, sessionmaker from netmover.db import session_scope -from netmover.models import CopiedFile, FileFailure, utcnow +from netmover.models import ArchivedFile, CopiedFile, FileFailure, utcnow CopyKey = tuple[str, int, int] # (relpath, size, mtime_ns) @@ -72,6 +72,53 @@ class Ledger: s.execute(delete(CopiedFile).where(CopiedFile.id.in_(stale))) return len(stale) + # -- Archiv ------------------------------------------------------------------------ + + def archived_keys(self, rule_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 + ) + ).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 + ) -> 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, + ) + ) + if exists is None: + s.add( + ArchivedFile( + rule_id=rule_id, + src_relpath=relpath, + size=size, + mtime_ns=mtime_ns, + archive_path=archive_path, + ) + ) + + def prune_archived(self, rule_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 + ) + ).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))) + return len(stale) + def failures(self, rule_id: int) -> dict[str, FailureInfo]: with session_scope(self.session_factory) as s: rows = s.scalars(select(FileFailure).where(FileFailure.rule_id == rule_id)).all() diff --git a/src/netmover/mover/runner.py b/src/netmover/mover/runner.py index e5ea375..521aa62 100644 --- a/src/netmover/mover/runner.py +++ b/src/netmover/mover/runner.py @@ -56,6 +56,11 @@ 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 = "" def parse_owner(value: str | None) -> tuple[int, int] | None: @@ -92,6 +97,7 @@ 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, @@ -109,6 +115,11 @@ 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, ) @@ -176,24 +187,52 @@ class RuleRunner: 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) 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._dest_factory(plan, dst_mp, src_root) + make_dest = self._target( + "dst_path", + plan.dst_mount_id, + plan.dst_type, + plan.dst_name, + plan.dst_path, + dst_mp, + src_root, + ) + make_archive = ( + self._target( + "archive_path", + plan.archive_mount_id, + plan.archive_type, + plan.archive_name, + plan.archive_path, + arch_mp, + src_root, + ) + if plan.archive_kind != "none" + else None + ) except (ValidationError, TransferError, SftpError) as exc: return RunResult(RunStatus.ERROR, dry, message=str(exc)) def work(token: CancelToken) -> RunResult: - source = make_source() + opened: list[Source | Destination] = [] try: + source = make_source() + opened.append(source) dest = make_dest() - except BaseException: - source.close() - raise - try: + 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( plan.snapshot, source, @@ -201,10 +240,11 @@ class RuleRunner: token, dry_run=dry, quarantine_rel=plan.quarantine_path, + archive=archive, ) finally: - dest.close() - source.close() + for item in reversed(opened): + item.close() def finish() -> None: self._started.pop(rule_id, None) @@ -251,41 +291,50 @@ class RuleRunner: return make_local - def _dest_factory( - self, plan: RulePlan, dst_mp: Path | None, src_root: Path | None + def _target( + self, + field: str, + mount_id: int | None, + mount_type: MountType | None, + mount_name: str | None, + path: str, + mp: Path | None, + src_root: Path | None, ) -> Callable[[], Destination]: - if plan.dst_mount_id is None: - root = Path(safe_abspath("dst_path", plan.dst_path)) + """Ziel oder Archiv: lokaler Pfad (``mount_id`` None), SFTP oder Kernel-Mount.""" + what = "Archiv" if field == "archive_path" else "Ziel" + if mount_id is None: + root = Path(safe_abspath(field, 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, " + field, + f"Lokales {what} liegt unter {base} – bitte die Anbindung als {what} 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) + elif mount_type is MountType.SFTP: + scfg = self.mounts.sftp_config_for(mount_id) + sub = safe_relpath(field, path) remote = posixpath.join(scfg.base_path, sub) if sub else scfg.base_path - label = plan.dst_name or "sftp" + label = mount_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" + if mp is None: + raise TransferError(f"{what}-Anbindung ist nicht eingehängt") + sub = safe_relpath(field, path) + local_root = mp / sub if sub else mp + kind = "cifs" if mount_type is MountType.CIFS else "nfs" create = True if src_root is not None and local_root.resolve() == src_root.resolve(): - raise ValidationError("dst_path", "Quelle und Ziel sind derselbe Ordner") + raise ValidationError(field, f"Quelle und {what} sind derselbe Ordner") def make_local() -> Destination: return LocalDestination(local_root, kind, create_root=create) # type: ignore[arg-type] diff --git a/tests/test_archive.py b/tests/test_archive.py new file mode 100644 index 0000000..2f3b289 --- /dev/null +++ b/tests/test_archive.py @@ -0,0 +1,351 @@ +"""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"