2 Commits
Author SHA1 Message Date
Florian WildeandClaude Opus 5.5 660bb70c5d feat(web): Archiv im Regelformular, Export/Import; Doku 0.7.0
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 00:22:39 +02:00
Florian WildeandClaude Opus 5.5 e59932ab75 feat: Archiv je Regel – erst Archiv, dann Ziel, dann Quelle löschen
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 <noreply@anthropic.com>
2026-10-12 00:22:39 +02:00
16 changed files with 854 additions and 43 deletions
+15 -1
View File
@@ -8,6 +8,19 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
## [Unreleased]
## [0.7.0] – 2026-10-12
### Hinzugefügt
- **Archiv je Regel:** zusätzliche Kopie in ein Archiv (Anbindung SMB/NFS/SFTP oder
lokaler Pfad), Ablage „wie in der Quelle“ oder „nach Datum“ (`Jahr/Monat/…`).
Reihenfolge: erst Archiv, dann Ziel, **erst dann wird die Quelle gelöscht**.
Scheitert das Ziel nach dem Archiv, verhindert eine Buchführung
(`archived_files`) eine zweite Archivkopie. Im Archiv wird nie überschrieben.
Nicht erreichbares Archiv → Regel wartet. Platzprüfung auch fürs Archiv,
Probelauf zeigt das Archivziel. Migration 0004.
- Export/Import enthalten die Archiv-Einstellungen; eine als Archiv benutzte
Anbindung lässt sich nicht löschen.
## [0.6.0] – 2026-10-11
### Geändert
@@ -246,7 +259,8 @@ Erste Version.
- Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.
[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.6.0...HEAD
[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.7.0...HEAD
[0.7.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.6.0...v0.7.0
[0.6.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.5.0...v0.6.0
[0.5.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.4.0...v0.5.0
[0.4.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.3.0...v0.4.0
+23 -1
View File
@@ -20,10 +20,32 @@ Zeiten in Ortszeit (Europe/Berlin).
| 2026-10-11 | 00:31 | 00:50 | 0:19 | – | Pausieren, Wartungsmodus, E-Mail (0.4.0) |
| 2026-10-11 | 00:54 | 01:16 | 0:22 | – | SFTP-Quelle, Zugangsdaten, Anmeldung (0.5.0) |
| 2026-10-11 | 08:43 | 08:55 | 0:12 | – | Anmeldung je Netz (0.6.0) |
| **Summe** | | | **3:29** | | |
| 2026-10-12 | 00:14 | 00:34 | 0:20 | – | Archiv je Regel (0.7.0) |
| **Summe** | | | **3:49** | | |
---
## 2026-10-12 · 00:14–00:34 · Archiv je Regel (0.7.0)
**Erledigt**
- Archivfelder an der Regel, Tabelle `archived_files` (Migration 0004).
- Engine: Archiv zuerst, dann Ziel, dann Quelle löschen; Archiv immer „umbenennen“.
- Runner: Ziel- und Archivaufbau über eine gemeinsame Fabrik (`_target`).
- 14 Tests (350 gesamt); Mutationsprobe 4/4 erkannt (Buchführung, Archiv
übersprungen, Kollisionsregel, Platzprüfung).
**Entscheidungen**
- Archiv *vor* Ziel: Paperless holt Dateien sofort ab – andersherum würde ein
gescheitertes Archiv beim Neuversuch zur doppelten Lieferung führen.
- Archiv überschreibt nie (unabhängig von der Kollisionsregel des Ziels).
- Ablage „nach Datum“ nach Verarbeitungsdatum (nicht Datei-mtime) – eindeutig
nachvollziehbar, wann eine Datei durchgelaufen ist.
**Stolperstein**
- Downgrade von 0004 scheiterte an den CHECK-Regeln der neuen Auswahlfelder
(verweisen nach dem Entfernen der Spalten ins Leere) – nur dank der Migrations-
tests mit Daten aufgefallen; Downgrade entfernt sie jetzt zuerst.
## 2026-10-11 · 08:43–08:55 · Anmeldung je Netz (0.6.0)
**Anlass** – Wunsch: Anmeldung abschalten, und zwar konfigurierbar, aus welchen Netzen
+1
View File
@@ -28,6 +28,7 @@ Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht
| Version | Thema |
|---|---|
| 0.7.0 | Archiv je Regel: Kopie ins Ziel und ins Archiv, Quelle erst danach löschen |
| 0.6.0 | Anmeldung je Netz: ohne Passwort / mit Passwort / abgewiesen, Schutz vor Aussperren |
| 0.5.0 | SFTP als Quelle; gespeicherte Zugangsdaten für SMB/SFTP-Passwörter (Seite „Zugangsdaten“); Anmeldung abschaltbar (vertrauenswürdige Netze / ganz) |
| 0.4.0 | Automatisch pausieren/fortsetzen bei nicht erreichbaren Anbindungen; Wartungsmodus; E-Mail-Benachrichtigung; eine Meldung pro ausgefallener Anbindung |
+1 -1
View File
@@ -4,7 +4,7 @@
services:
netmover:
build: .
image: netmover:0.6.0
image: netmover:0.7.0
container_name: netmover
restart: unless-stopped
cap_add:
+31 -2
View File
@@ -8,7 +8,7 @@ Diese Anleitung erklärt die Bedienung. Installation, Updates und technische Det
stehen im [README](../README.md).
**Inhalt:** [Begriffe](#begriffe) · [Erste Anmeldung](#erste-anmeldung) ·
[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) ·
[Zugangsdaten](#zugangsdaten) · [Anbindungen](#anbindungen) · [Regeln](#regeln) · [Archiv](#archiv) ·
[Probelauf](#probelauf) · [Status](#status) · [Wartungsmodus](#wartungsmodus) · [Protokoll](#protokoll) ·
[Einstellungen](#einstellungen) · [Anmeldung](#anmeldung) · [Uptime Kuma](#uptime-kuma) · [Beispiele](#beispiele) · [Häufige Fragen](#fragen)
@@ -160,6 +160,34 @@ Gespeicherte Zugangsdaten lassen sich nur löschen, wenn sie keine Anbindung meh
| **Aktion** | **verschieben** (Quelle wird nach erfolgreicher Übertragung gelöscht) oder **kopieren** (Quelle bleibt; jede Dateiversion wird nur einmal kopiert). |
| **Wenn Datei existiert** | **umbenennen** (`scan.pdf` → `scan-1.pdf`, `scan-2.pdf` …), **überspringen** oder **überschreiben**. |
### Archiv {#archiv}
Optional bekommt jede Datei zusätzlich eine **Kopie ins Archiv** – z. B. Paperless als
Ziel und ein Archivordner auf dem NAS. **Die Quelle wird erst gelöscht, wenn die Datei
im Ziel UND im Archiv angekommen ist.**
| Feld | Bedeutung |
|---|---|
| **Archiv** | kein Archiv / eine Anbindung (SMB, NFS, SFTP) / ein lokaler Pfad |
| **Archivordner** | Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren. |
| **Ablage** | **wie in der Quelle** oder **nach Datum** – `2026/10/scan001.pdf` (Monat der Verarbeitung) |
So läuft es pro Datei:
1. **Zuerst ins Archiv** – mit denselben Sicherungen wie beim Ziel.
2. **Dann ins Ziel.**
3. **Dann erst** wird die Quelle gelöscht (bei „verschieben“).
Das Archiv kommt zuerst, weil ein Ziel wie Paperless Dateien sofort abholt: Scheitert
das Archiv, wurde noch nichts geliefert, und die Datei bleibt in der Quelle. Scheitert
danach das Ziel, merkt sich NetMover, dass die Datei schon archiviert ist – der nächste
Versuch legt keine zweite Archivkopie an.
Im Archiv wird **nie überschrieben**: Gleiche Namen werden immer umbenannt
(`scan-1.pdf`), egal was die Regel fürs Ziel vorsieht. Ist das Archiv nicht erreichbar,
wartet die Regel wie bei Quelle und Ziel. Im Kopiermodus bekommen Ziel und Archiv je
eine Kopie, die Quelle bleibt.
### Sicherheit und Grenzen {#regel-sicherheit}
| Feld | Bedeutung | Empfehlung |
@@ -378,7 +406,8 @@ deaktiviert, Zugangsdaten müssen neu eingegeben werden.
3. Anbindung **Paperless** (SFTP, Pfad `/opt/paperless_data/consume`, Schlüssel
wählen, Benutzer = der Paperless-Benutzer), testen.
4. Regel **Scans → Paperless**: Quelle NAS, Filter `*.pdf`, Ziel Paperless,
verschieben, umbenennen, 30 s. **Probelauf**, dann aktivieren.
verschieben, umbenennen, 30 s. Optional **Archiv**: Anbindung NAS, Ordner
`Archiv/Scans`, Ablage nach Datum. **Probelauf**, dann aktivieren.
### Jede Person in einen eigenen Ordner {#beispiel-unterordner}
+1 -1
View File
@@ -1,3 +1,3 @@
"""NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
__version__ = "0.6.0"
__version__ = "0.7.0"
+52 -1
View File
@@ -18,6 +18,8 @@ from sqlalchemy.orm import Session
from netmover.crypto import SecretBox, generate_ssh_keypair
from netmover.models import (
ArchiveKind,
ArchiveLayout,
CollisionPolicy,
Credential,
CredentialKind,
@@ -424,7 +426,11 @@ def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None:
if m is None:
return
used = s.scalars(
select(Rule.name).where((Rule.src_mount_id == mount_id) | (Rule.dst_mount_id == mount_id))
select(Rule.name).where(
(Rule.src_mount_id == mount_id)
| (Rule.dst_mount_id == mount_id)
| (Rule.archive_mount_id == mount_id)
)
).all()
if used:
raise AdminError(f"Wird noch von Regeln benutzt: {', '.join(used)}")
@@ -465,6 +471,10 @@ class RuleInput:
dry_run: bool = False
quarantine_path: str = ""
quarantine_after: int = 3
archive_kind: ArchiveKind = ArchiveKind.NONE
archive_mount_id: int | None = None
archive_path: str = ""
archive_layout: ArchiveLayout = ArchiveLayout.FLAT
extra: dict[str, str] = field(default_factory=dict)
@@ -492,6 +502,10 @@ def rule_input_from_model(r: Rule) -> RuleInput:
dry_run=r.dry_run,
quarantine_path=r.quarantine_path or "",
quarantine_after=r.quarantine_after,
archive_kind=r.archive_kind,
archive_mount_id=r.archive_mount_id,
archive_path=r.archive_path,
archive_layout=r.archive_layout,
)
@@ -585,6 +599,39 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
d.quarantine_path = safe_relpath("quarantine_path", d.quarantine_path)
except ValidationError as exc:
errors["quarantine_path"] = str(exc)
# Archiv
try:
d.archive_kind = ArchiveKind(form.get("archive_kind") or "none")
d.archive_layout = ArchiveLayout(form.get("archive_layout") or "flat")
except ValueError:
errors["archive_kind"] = "Ungültige Auswahl."
d.archive_path = (form.get("archive_path") or "").strip()
if d.archive_kind is ArchiveKind.MOUNT:
d.archive_mount_id = _int(form, "archive_mount_id", errors)
if d.archive_mount_id is None or s.get(Mount, d.archive_mount_id) is None:
errors["archive_mount_id"] = "Archiv-Anbindung wählen."
try:
d.archive_path = safe_relpath("archive_path", d.archive_path)
except ValidationError as exc:
errors["archive_path"] = str(exc)
if (
"archive_path" not in errors
and d.archive_mount_id == d.dst_mount_id
and d.dst_kind == "mount"
and d.archive_path == d.dst_path
):
errors["archive_path"] = "Archiv und Ziel sind derselbe Ordner."
elif d.archive_kind is ArchiveKind.LOCAL:
d.archive_mount_id = None
try:
d.archive_path = safe_abspath("archive_path", d.archive_path)
except ValidationError as exc:
errors["archive_path"] = str(exc)
if d.dst_kind == "local" and d.archive_path == d.dst_path:
errors["archive_path"] = "Archiv und Ziel sind derselbe Ordner."
else:
d.archive_mount_id, d.archive_path = None, ""
return d, errors
@@ -617,6 +664,10 @@ def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rul
"on_collision",
"dry_run",
"quarantine_after",
"archive_kind",
"archive_mount_id",
"archive_path",
"archive_layout",
):
setattr(r, f, getattr(d, f))
r.set_owner = d.set_owner or None
@@ -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")
+41
View File
@@ -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."""
+85 -9
View File
@@ -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:
+48 -1
View File
@@ -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()
+73 -24
View File
@@ -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:
opened: list[Source | Destination] = []
try:
source = make_source()
try:
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]
+10 -1
View File
@@ -16,6 +16,8 @@ from sqlalchemy.orm import Session
from netmover import __version__
from netmover.models import (
ArchiveKind,
ArchiveLayout,
CollisionPolicy,
Mount,
MountState,
@@ -72,6 +74,9 @@ RULE_FIELDS = (
"dry_run",
"quarantine_path",
"quarantine_after",
"archive_kind",
"archive_path",
"archive_layout",
)
SETTING_FIELDS = (
"retention_days",
@@ -99,6 +104,7 @@ def export_config(s: Session, settings: AppSettings) -> dict[str, Any]:
**{f: _plain(getattr(r, f)) for f in RULE_FIELDS},
"src_mount": names[r.src_mount_id],
"dst_mount": names.get(r.dst_mount_id) if r.dst_mount_id else None,
"archive_mount": names.get(r.archive_mount_id) if r.archive_mount_id else None,
}
for r in s.scalars(select(Rule).order_by(Rule.name))
],
@@ -165,13 +171,16 @@ def import_config(s: Session, mounts: MountManager, data: dict[str, Any]) -> Imp
try:
src = ids[raw["src_mount"]]
dst = ids[raw["dst_mount"]] if raw.get("dst_mount") else None
r = Rule(src_mount_id=src, dst_mount_id=dst)
arch = ids[raw["archive_mount"]] if raw.get("archive_mount") else None
r = Rule(src_mount_id=src, dst_mount_id=dst, archive_mount_id=arch)
for f in RULE_FIELDS:
if f in raw:
setattr(r, f, raw[f])
r.pattern_type = PatternType(r.pattern_type)
r.action = RuleAction(r.action)
r.on_collision = CollisionPolicy(r.on_collision)
r.archive_kind = ArchiveKind(r.archive_kind or "none")
r.archive_layout = ArchiveLayout(r.archive_layout or "flat")
r.src_subpath = safe_relpath("src_subpath", r.src_subpath or "")
if int(r.interval_seconds) < 10:
raise ValueError("Intervall < 10 s")
+12 -1
View File
@@ -422,13 +422,24 @@ class RuleView:
target: str
def _archive_target(r: Rule) -> str | None:
if r.archive_kind.value == "none":
return None
if r.archive_kind.value == "local" or r.archive_mount is None:
place = r.archive_path
else:
place = f"{r.archive_mount.name}:/{r.archive_path}"
return place + (" (nach Datum)" if r.archive_layout.value == "date" else "")
def _rule_target(r: Rule) -> str:
src = f"{r.src_mount.name}:/{r.src_subpath}" if r.src_subpath else f"{r.src_mount.name}:/"
if r.dst_mount is None:
dst = r.dst_path
else:
dst = f"{r.dst_mount.name}:/{r.dst_path}" if r.dst_path else f"{r.dst_mount.name}:/"
return f"{src} → {dst}"
archive = _archive_target(r)
return f"{src} → {dst}" + (f" + Archiv {archive}" if archive else "")
async def _status_context(request: Request) -> dict[str, Any]:
@@ -30,6 +30,13 @@
{{ select('on_collision', 'Wenn Datei existiert', [('rename', 'umbenennen (scan-1.pdf)'), ('skip', 'überspringen'), ('overwrite', 'überschreiben')], d.on_collision.value, errors=errors) }}
</div></fieldset>
<fieldset><legend>Archiv <a class="legend-help" href="/help#archiv" title="Hilfe">?</a></legend><div class="fields">
{{ select('archive_kind', 'Archiv', [('none', 'kein Archiv'), ('mount', 'Anbindung'), ('local', 'Lokaler Pfad')], d.archive_kind.value, errors=errors, hint='Zusätzliche Kopie. Die Quelle wird erst gelöscht, wenn Ziel UND Archiv sie haben.') }}
<div data-show-for="archive_kind:mount">{{ select('archive_mount_id', 'Archiv-Anbindung', dst_options, d.archive_mount_id or '', errors=errors) }}</div>
<div data-show-for="archive_kind:mount local">{{ field('archive_path', 'Archivordner', d.archive_path, errors=errors, hint='Anbindung: relativer Unterordner (wird angelegt). Lokal: absoluter Pfad, muss existieren.') }}</div>
<div data-show-for="archive_kind:mount local">{{ select('archive_layout', 'Ablage', [('flat', 'wie in der Quelle'), ('date', 'nach Datum (Jahr/Monat)')], d.archive_layout.value, errors=errors, hint='Im Archiv wird nie überschrieben – gleiche Namen werden umbenannt.') }}</div>
</div></fieldset>
<fieldset><legend>Sicherheit und Grenzen <a class="legend-help" href="/help#regel-sicherheit" title="Hilfe">?</a></legend><div class="fields">
{{ field('stability_seconds', 'Stabilitätsfenster (s)', d.stability_seconds, type='number', errors=errors, attrs='min="0"', hint='Größe und Zeit müssen so lange gleich bleiben') }}
{{ field('min_age_seconds', 'Mindestalter (s)', d.min_age_seconds, type='number', errors=errors, attrs='min="0"') }}
+351
View File
@@ -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"