From b6b6bf6bd7dfafb3de8441a5fe7085b87fccab10 Mon Sep 17 00:00:00 2001 From: Florian Wilde Date: Sun, 11 Oct 2026 00:47:02 +0200 Subject: [PATCH] feat: Regeln pausieren automatisch, solange eine Anbindung fehlt; Wartungsmodus MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wartezustand mit einem Protokolleintrag statt eines Fehlers alle 30 s, automatisches Fortsetzen; Meldung einmal pro Anbindung mit betroffenen Regeln (auch beim Aushängen von außen). Migration 0003 mit Tests auf befüllten Datenbanken. Co-Authored-By: Claude Opus 5.5 --- src/netmover/health.py | 19 +- .../versions/0003_rule_waiting_state.py | 32 +++ src/netmover/models.py | 5 + src/netmover/mounting/manager.py | 45 +++-- src/netmover/mover/engine.py | 2 + src/netmover/mover/runner.py | 2 +- src/netmover/runs.py | 42 ++++ tests/test_migrations.py | 45 ++++- tests/test_waiting.py | 184 ++++++++++++++++++ 9 files changed, 359 insertions(+), 17 deletions(-) create mode 100644 src/netmover/migrations/versions/0003_rule_waiting_state.py create mode 100644 tests/test_waiting.py diff --git a/src/netmover/health.py b/src/netmover/health.py index 3dbcb4b..40c1ed2 100644 --- a/src/netmover/health.py +++ b/src/netmover/health.py @@ -49,6 +49,7 @@ class Health: problems: list[str] = field(default_factory=list) stale_rules: set[int] = field(default_factory=set) checks: list[Check] = field(default_factory=list) + maintenance: bool = False def add(self, check: Check) -> None: self.checks.append(check) @@ -86,12 +87,22 @@ def _mount_check(m: Mount) -> Check: return mk(True, f"Anbindung „{m.name}“ in Ordnung") -def _rule_check(r: Rule, service_started: datetime, now: datetime) -> Check: +def _rule_check( + r: Rule, service_started: datetime, now: datetime, maintenance: bool = False +) -> Check: def mk(ok: bool, message: str) -> Check: return Check("rule", r.name, ok, message, id=r.id) if not r.enabled: return mk(True, f"Regel „{r.name}“ deaktiviert") + if maintenance: + return mk(True, f"Regel „{r.name}“ pausiert (Wartungsmodus)") + if r.waiting_since is not None: + return mk( + False, + f"Regel „{r.name}“ wartet seit {r.waiting_since:%Y-%m-%d %H:%M} UTC auf eine " + f"Anbindung: {r.waiting_reason or ''}".strip(), + ) if rule_is_stale(r, service_started, now): ago = ( "noch nie" @@ -113,9 +124,11 @@ def check_health( last_heartbeat: datetime | None, service_started: datetime, now: datetime | None = None, + maintenance: bool = False, ) -> Health: now = now or utcnow() health = Health() + health.maintenance = maintenance if not scheduler_running: health.add(Check("scheduler", "Scheduler", False, "Scheduler läuft nicht")) elif last_heartbeat is None or now - last_heartbeat > HEARTBEAT_MAX_AGE: @@ -128,8 +141,8 @@ def check_health( for m in s.scalars(select(Mount).order_by(Mount.name)): health.add(_mount_check(m)) for r in s.scalars(select(Rule).order_by(Rule.name)): - check = _rule_check(r, service_started, now) - if r.enabled and rule_is_stale(r, service_started, now): + check = _rule_check(r, service_started, now, maintenance) + if r.enabled and not maintenance and rule_is_stale(r, service_started, now): health.stale_rules.add(r.id) health.add(check) return health diff --git a/src/netmover/migrations/versions/0003_rule_waiting_state.py b/src/netmover/migrations/versions/0003_rule_waiting_state.py new file mode 100644 index 0000000..7c8faa3 --- /dev/null +++ b/src/netmover/migrations/versions/0003_rule_waiting_state.py @@ -0,0 +1,32 @@ +"""rule waiting state + +Revision ID: 0003 +Revises: 0002 +Create Date: 2026-10-11 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +import netmover.models + +revision: str = "0003" +down_revision: str | None = "0002" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + with op.batch_alter_table("rules", schema=None) as batch_op: + batch_op.add_column( + sa.Column("waiting_since", netmover.models.UTCDateTime(), nullable=True) + ) + batch_op.add_column(sa.Column("waiting_reason", sa.Text(), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("rules", schema=None) as batch_op: + batch_op.drop_column("waiting_reason") + batch_op.drop_column("waiting_since") diff --git a/src/netmover/models.py b/src/netmover/models.py index 4811b5e..2d6d96d 100644 --- a/src/netmover/models.py +++ b/src/netmover/models.py @@ -10,6 +10,8 @@ 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.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. * ``mounts.write_ok`` / ``mounts.root_squash`` / ``mounts.last_test_*`` – Ergebnis des letzten Mount-Tests; ``root_squash`` entscheidet, ob ``chown`` auf NFS-Zielen @@ -260,6 +262,9 @@ class Rule(Base): 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) + # Automatisch pausiert, solange Quelle oder Ziel nicht erreichbar ist (Migration 0003) + waiting_since: Mapped[datetime | None] = mapped_column(UTCDateTime) + waiting_reason: Mapped[str | None] = mapped_column(Text) created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) src_mount: Mapped[Mount] = relationship(foreign_keys=[src_mount_id]) diff --git a/src/netmover/mounting/manager.py b/src/netmover/mounting/manager.py index 64bd8e6..7af6ff4 100644 --- a/src/netmover/mounting/manager.py +++ b/src/netmover/mounting/manager.py @@ -32,7 +32,7 @@ from sqlalchemy.orm import Session, sessionmaker from netmover.crypto import SecretBox, redact from netmover.db import session_scope from netmover.logging_setup import event, get_logger -from netmover.models import CredentialKind, Mount, MountState, MountType, utcnow +from netmover.models import CredentialKind, Mount, MountState, MountType, Rule, utcnow from netmover.mounting.backend import CifsCredentials, MountBackend, MountError, MountSpec from netmover.mounting.report import CheckReport, assess_write from netmover.mounting.sftp import SftpConfig, SftpError, check_connection, connect @@ -203,12 +203,31 @@ class MountManager: with self._locks_guard: return self._locks.setdefault(mount_id, threading.RLock()) - def _notify(self, kind: str, name: str, message: str) -> None: - if self._notifier is not None: - try: - self._notifier(kind, name, message) - except Exception: - log.exception("notifier.failed") + def _notify(self, kind: str, name: str, message: str, mount_id: int | None = None) -> None: + if self._notifier is None: + return + if mount_id is not None: + rules = self._affected_rules(mount_id) + if rules and kind != "mount.recovered": + waiting = "\nBetroffene Regeln (warten, bis die Anbindung zurück ist): " + message += waiting + ", ".join(rules) + elif rules: + message += f"\nRegeln laufen wieder: {', '.join(rules)}" + try: + self._notifier(kind, name, message) + except Exception: + log.exception("notifier.failed") + + def _affected_rules(self, mount_id: int) -> list[str]: + with session_scope(self.session_factory) as s: + return list( + s.scalars( + select(Rule.name) + .where(Rule.enabled.is_(True)) + .where((Rule.src_mount_id == mount_id) | (Rule.dst_mount_id == mount_id)) + .order_by(Rule.name) + ) + ) def mountpoint_for(self, slug: str) -> Path: if slugify(slug) != slug: @@ -347,7 +366,7 @@ class MountManager: # z. B. Stromausfall: die VM bootet schneller als das NAS self._retry_at[mount_id] = self._clock() + self.retry_interval results.append(RestoreResult(name, False, str(exc))) - self._notify("mount.restore_failed", name, str(exc)) + self._notify("mount.restore_failed", name, str(exc), mount_id) for r in results: event( log, @@ -531,6 +550,8 @@ class MountManager: self._retry_mount(mount_id, m, cred, mp) return mp msg = f"„{m.name}“ ist nicht eingehängt – Lauf abgebrochen, nichts angefasst." + if m.state is not MountState.ERROR: + self._notify("mount.error", m.name, msg, mount_id) self._set_state(mount_id, MountState.ERROR, msg) raise MountUnavailable(msg) try: @@ -563,8 +584,10 @@ class MountManager: msg = f"„{m.name}“ nicht erreichbar: {exc}" self._set_state(mount_id, MountState.ERROR, msg) if m.state is not MountState.ERROR: - self._notify("mount.error", m.name, msg) + self._notify("mount.error", m.name, msg, mount_id) raise MountUnavailable(msg) from None + if m.state is MountState.ERROR: + self._notify("mount.recovered", m.name, f"„{m.name}“ ist wieder erreichbar.", mount_id) if m.state is not MountState.MOUNTED: self._set_state(mount_id, MountState.MOUNTED) @@ -601,7 +624,7 @@ class MountManager: self._retry_at[mount_id] = self._clock() + self.retry_interval msg += f"\n→ Neuer Versuch automatisch alle {self.retry_interval:g} s." self._set_state(mount_id, MountState.ERROR, msg) - self._notify("mount.error", m.name, msg) + self._notify("mount.error", m.name, msg, mount_id) raise MountUnavailable(msg) def retry_pending(self) -> list[str]: @@ -663,7 +686,7 @@ class MountManager: del self._retry_at[mount_id] self._set_state(mount_id, MountState.MOUNTED, mounted=True) event(log, "mount.recovered", mount=m.name, attempt="retry") - self._notify("mount.recovered", m.name, f"„{m.name}“ ist wieder eingehängt.") + self._notify("mount.recovered", m.name, f"„{m.name}“ ist wieder eingehängt.", mount_id) def usage(self, mount_id: int, timeout: float | None = None) -> tuple[int, int] | None: """(gesamt, frei) in Bytes – nur bei eingehängten Kernel-Mounts.""" diff --git a/src/netmover/mover/engine.py b/src/netmover/mover/engine.py index 85ca126..1b45ced 100644 --- a/src/netmover/mover/engine.py +++ b/src/netmover/mover/engine.py @@ -92,6 +92,8 @@ class RunResult: failed: int = 0 message: str | None = None items: list[ItemResult] = field(default_factory=list) + # Quelle oder Ziel nicht erreichbar → Regel wartet (automatisch pausiert) + unavailable: bool = False def add(self, item: ItemResult) -> None: self.items.append(item) diff --git a/src/netmover/mover/runner.py b/src/netmover/mover/runner.py index cb96dfa..e36d6c4 100644 --- a/src/netmover/mover/runner.py +++ b/src/netmover/mover/runner.py @@ -178,7 +178,7 @@ class RuleRunner: else None ) except MountUnavailable as exc: - return RunResult(RunStatus.ERROR, dry, message=str(exc)) + return RunResult(RunStatus.ERROR, dry, message=str(exc), unavailable=True) if src_mp is None: # nur bei SFTP, oben bereits ausgeschlossen return RunResult(RunStatus.ERROR, dry, message="Quelle ist kein Kernel-Mount") src_root = src_mp / plan.src_subpath if plan.src_subpath else src_mp diff --git a/src/netmover/runs.py b/src/netmover/runs.py index 7cc4d66..8e1d829 100644 --- a/src/netmover/runs.py +++ b/src/netmover/runs.py @@ -8,6 +8,14 @@ die Einzeldateien in ``run_items``. Danach die Regelzähler: * ``consecutive_failures`` – ``error`` und ``partial`` zählen hoch, ``ok`` setzt zurück. Ein manuell gestarteter Probelauf verändert keine Zähler. + +Automatisch pausieren (0.4.0): Ist Quelle oder Ziel nicht erreichbar, wechselt die +Regel in „wartet auf Anbindung“. Protokolliert wird nur der Beginn, nicht jeder +weitere Termin, und „Fehler in Folge“ zählt nicht hoch – die Benachrichtigung +kommt einmal pro Anbindung vom MountManager. Sobald ein Lauf wieder durchkommt, +endet das Warten von selbst, und der Lauf vermerkt die Wartezeit. + +Wartungsmodus: geplante Läufe werden gar nicht erst ausgeführt. """ from __future__ import annotations @@ -51,6 +59,8 @@ class RunService: ) -> int | None: """Führt die Regel aus, protokolliert den Lauf und liefert die Run-ID (None, wenn die Regel inzwischen gelöscht wurde).""" + if trigger is RunTrigger.SCHEDULE and load_settings(self.session_factory).maintenance: + return None started = utcnow() result = self.runner.run(rule_id, dry_run=dry_run) finished = utcnow() @@ -73,6 +83,29 @@ class RunService: if rule is None: return None message = result.message + if update_counters and result.unavailable: + if rule.waiting_since is not None: + # wartet bereits – nur den Grund aktualisieren, kein neuer Eintrag + rule.waiting_reason = result.message + event(log, "rule.still_waiting", level=10, rule=rule.name) + return None + rule.waiting_since = finished + rule.waiting_reason = result.message + rule.last_run_at = finished + message = f"Pausiert, bis die Anbindung zurück ist: {result.message}" + update_counters = False # „Fehler in Folge“ nicht hochzählen + event(log, "rule.waiting", level=30, rule=rule.name, reason=result.message or "") + elif ( + update_counters + and rule.waiting_since is not None + and result.status is not RunStatus.SKIPPED + ): + waited = _human(finished - rule.waiting_since) + rule.waiting_since = None + rule.waiting_reason = None + note = f"Fortgesetzt nach {waited} Wartezeit" + message = f"{note}; {message}" if message else note + event(log, "rule.resumed", rule=rule.name, waited=waited) items = result.items if len(items) > MAX_ITEMS_PER_RUN: extra = len(items) - MAX_ITEMS_PER_RUN @@ -153,6 +186,15 @@ class RunService: return "\n".join(lines) +def _human(delta: timedelta) -> str: + seconds = int(delta.total_seconds()) + if seconds < 90: + return f"{seconds} s" + if seconds < 5400: + return f"{round(seconds / 60)} min" + return f"{seconds / 3600:.1f} h".replace(".", ",") + + def cleanup(session_factory: sessionmaker[Session], now: datetime | None = None) -> dict[str, int]: """Protokolle älter als die Aufbewahrungsfrist löschen (Items per CASCADE), dazu Fehlerzähler, die seitdem nicht mehr angefasst wurden.""" diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 26c5d6b..1b34143 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -18,6 +18,14 @@ from alembic import command from netmover.db import alembic_config, upgrade +def _head() -> str: + from alembic.script import ScriptDirectory + + head = ScriptDirectory.from_config(alembic_config("sqlite://")).get_current_head() + assert head is not None + return head + + def _seed(db: Path) -> None: """Datenbank auf Stand 0001 mit verknüpften Daten, wie auf einem echten System.""" c = sqlite3.connect(db) @@ -80,7 +88,7 @@ def test_upgrade_with_linked_data(db_0001: tuple[str, Path]) -> None: url, db = db_0001 upgrade(url) st = _state(db) - assert st["version"] == "0002" + assert st["version"] == _head() assert st["tmp_tables"] == [] and st["fk_violations"] == [] assert "managed" in st["cred_columns"] # type: ignore[operator] assert st["mounts"] == [("NAS", 1), ("paperless", 2)] @@ -122,7 +130,7 @@ def test_failed_migration_leaves_database_untouched( monkeypatch.setattr(DefaultImpl, "drop_table", real_drop) upgrade(url) # danach klappt es ohne Handarbeit - assert _state(db)["version"] == "0002" + assert _state(db)["version"] == _head() def test_app_engine_still_enforces_foreign_keys(db_url: str) -> None: @@ -134,3 +142,36 @@ def test_app_engine_still_enforces_foreign_keys(db_url: str) -> None: with engine.connect() as conn: assert conn.execute(text("PRAGMA foreign_keys")).scalar() == 1 engine.dispose() + + +def test_upgrade_from_0002_keeps_everything_linked_to_rules(tmp_path: Path) -> None: + """0003 baut ``rules`` neu auf – Läufe, Buchführung und Fehlerzähler verweisen darauf.""" + db = tmp_path / "v02.db" + url = f"sqlite:///{db}" + upgrade(url, "0001") + _seed(db) + upgrade(url, "0002") + c = sqlite3.connect(db) + c.execute("PRAGMA foreign_keys=ON") + c.executescript( + """ + UPDATE credentials SET managed = 1 WHERE id = 2; + UPDATE mounts SET username = 'paperless' WHERE id = 2; + INSERT INTO copied_files (rule_id, src_relpath, size, mtime_ns, dst_path, copied_at) + VALUES (1, 'a.pdf', 1, 1, '/x/a.pdf', '2026-10-10 20:00:00'); + INSERT INTO file_failures (rule_id, src_relpath, size, mtime_ns, attempts, last_attempt_at) + VALUES (1, 'b.pdf', 1, 1, 2, '2026-10-10 20:00:00'); + """ + ) + c.commit() + c.close() + upgrade(url) + st = _state(db) + assert st["version"] == _head() and st["fk_violations"] == [] and st["tmp_tables"] == [] + assert st["rules"] == [("Scans → Paperless", 1, 2)] and st["run_items"] == 1 + c = sqlite3.connect(db) + assert c.execute("SELECT COUNT(*) FROM copied_files").fetchone()[0] == 1 + assert c.execute("SELECT attempts FROM file_failures").fetchone()[0] == 2 + assert c.execute("SELECT waiting_since FROM rules").fetchone()[0] is None + assert c.execute("SELECT username FROM mounts WHERE id = 2").fetchone()[0] == "paperless" + c.close() diff --git a/tests/test_waiting.py b/tests/test_waiting.py new file mode 100644 index 0000000..e536e2f --- /dev/null +++ b/tests/test_waiting.py @@ -0,0 +1,184 @@ +"""0.4.0: automatisch pausieren/fortsetzen, Wartungsmodus, eine Meldung pro Anbindung.""" + +from __future__ import annotations + +from datetime import timedelta +from pathlib import Path + +import pytest +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from netmover.db import session_scope +from netmover.health import check_health +from netmover.models import Rule, Run, RunStatus, RunTrigger, utcnow +from netmover.mounting.fake import FakeMountBackend +from netmover.mounting.manager import MountManager +from netmover.mover.engine import RunResult +from netmover.mover.runner import RuleRunner +from netmover.runs import RunService +from netmover.settings_store import AppSettings, save_settings +from tests.conftest import MountFactory + + +class FakeRunner: + def __init__(self) -> None: + self.results: list[RunResult] = [] + self.calls = 0 + + def run(self, rule_id: int, *, dry_run: bool | None = None) -> RunResult: + self.calls += 1 + return self.results.pop(0) + + +@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="Scans", src_mount_id=mid, dst_path="/x", interval_seconds=30) + s.add(r) + s.flush() + return r.id + + +def _rule(sf: sessionmaker[Session], rid: int) -> Rule: + with sf() as s: + r = s.get(Rule, rid) + assert r is not None + s.expunge(r) + return r + + +def _runs(sf: sessionmaker[Session]) -> list[Run]: + with sf() as s: + return list(s.scalars(select(Run).order_by(Run.id))) + + +def unavailable() -> RunResult: + return RunResult( + RunStatus.ERROR, + message="„NAS“ ist nicht eingehängt – Lauf abgebrochen, nichts angefasst.", + unavailable=True, + ) + + +def test_waiting_logs_once_and_resumes_by_itself( + session_factory: sessionmaker[Session], rule_id: int +) -> None: + runner = FakeRunner() + service = RunService(session_factory, runner) # type: ignore[arg-type] + runner.results += [unavailable(), unavailable(), unavailable(), RunResult(RunStatus.OK)] + + first = service.execute(rule_id) + assert first is not None + r = _rule(session_factory, rule_id) + assert r.waiting_since is not None and "nicht eingehängt" in (r.waiting_reason or "") + assert r.consecutive_failures == 0 # Warten ist kein Fehlschlag + + assert service.execute(rule_id) is None # kein neuer Eintrag alle 30 s + assert service.execute(rule_id) is None + assert len(_runs(session_factory)) == 1 + assert _runs(session_factory)[0].message.startswith("Pausiert, bis die Anbindung zurück ist") + + service.execute(rule_id) # Anbindung wieder da + r = _rule(session_factory, rule_id) + assert r.waiting_since is None and r.waiting_reason is None and r.last_success_at is not None + runs = _runs(session_factory) + assert len(runs) == 2 and runs[1].message and runs[1].message.startswith("Fortgesetzt nach") + assert r.consecutive_failures == 0 + + +def test_real_errors_still_count(session_factory: sessionmaker[Session], rule_id: int) -> None: + runner = FakeRunner() + service = RunService(session_factory, runner) # type: ignore[arg-type] + runner.results += [RunResult(RunStatus.ERROR, message="kaputt")] * 2 + service.execute(rule_id) + service.execute(rule_id) + r = _rule(session_factory, rule_id) + assert r.consecutive_failures == 2 and r.waiting_since is None + assert len(_runs(session_factory)) == 2 + + +def test_manual_dry_run_does_not_start_waiting( + session_factory: sessionmaker[Session], rule_id: int +) -> None: + runner = FakeRunner() + service = RunService(session_factory, runner) # type: ignore[arg-type] + runner.results.append(unavailable()) + service.execute(rule_id, RunTrigger.MANUAL, dry_run=True) + assert _rule(session_factory, rule_id).waiting_since is None + assert len(_runs(session_factory)) == 1 # Probelauf sieht man trotzdem + + +def test_maintenance_skips_scheduled_runs_but_not_manual( + session_factory: sessionmaker[Session], rule_id: int +) -> None: + runner = FakeRunner() + service = RunService(session_factory, runner) # type: ignore[arg-type] + save_settings(session_factory, AppSettings(maintenance=True)) + assert service.execute(rule_id) is None + assert runner.calls == 0 and _runs(session_factory) == [] + runner.results.append(RunResult(RunStatus.OK)) + assert service.execute(rule_id, RunTrigger.MANUAL) is not None + assert runner.calls == 1 + + +def _health(sf: sessionmaker[Session], **kw: object): # type: ignore[no-untyped-def] + now = utcnow() + return check_health( + sf, + scheduler_running=True, + last_heartbeat=now, + service_started=now - timedelta(hours=1), + now=now, + **kw, + ) # type: ignore[arg-type] + + +def test_health_waiting_and_maintenance( + session_factory: sessionmaker[Session], rule_id: int +) -> None: + with session_scope(session_factory) as s: + r = s.get(Rule, rule_id) + assert r is not None + r.created_at = utcnow() - timedelta(days=1) + r.last_success_at = utcnow() - timedelta(hours=2) + r.waiting_since = utcnow() - timedelta(minutes=12) + r.waiting_reason = "„NAS“ nicht verfügbar" + h = _health(session_factory) + assert not h.ok and "wartet seit" in h.problems[0] and "NAS" in h.problems[0] + h = _health(session_factory, maintenance=True) + assert h.ok and h.maintenance and h.stale_rules == set() + check = h.find("rule", id=rule_id) + assert check is not None and "Wartungsmodus" in check.message + + +# -- Integration mit MountManager und Runner ------------------------------------------------ + + +def test_vanished_mount_notifies_once_with_affected_rules( + manager: MountManager, + fake_backend: FakeMountBackend, + make_mount: MountFactory, + session_factory: sessionmaker[Session], + notifications: list[tuple[str, str, str]], + tmp_path: Path, +) -> None: + fake_backend.export("nas", "scans") + mid = make_mount("nas", share="scans") + manager.mount(mid) + with session_scope(session_factory) as s: + s.add(Rule(name="Scans → Paperless", src_mount_id=mid, dst_path=str(tmp_path))) + s.add(Rule(name="Archiv", src_mount_id=mid, dst_path=str(tmp_path), enabled=False)) + fake_backend.vanish(manager.mountpoint_for("nas")) + + runner = RuleRunner(session_factory, manager, sleep=lambda _: None) + with session_factory() as s: + rid = s.scalars(select(Rule.id).where(Rule.name == "Scans → Paperless")).one() + for _ in range(3): + result = runner.run(rid) + assert result.unavailable and result.status is RunStatus.ERROR + errors = [n for n in notifications if n[0] == "mount.error"] + assert len(errors) == 1 + assert "Betroffene Regeln" in errors[0][2] and "Scans → Paperless" in errors[0][2] + assert "Archiv" not in errors[0][2] # deaktivierte Regel nicht erwähnt