feat: Regeln pausieren automatisch, solange eine Anbindung fehlt; Wartungsmodus

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 <noreply@anthropic.com>

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()

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