diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ab369f..ed5e2b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,19 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`. Inaktivitäts-Zeitlimit; ein hängender Worker blockiert nur seine eigene Regel. - Lokale Ziele unter dem Mount-Basisverzeichnis werden abgelehnt (dort muss die Anbindung als Ziel gewählt werden, damit „eingehängt?“ geprüft wird). +- Scheduler (APScheduler): ein Job pro aktiver Regel mit zufälligem + Startversatz und Jitter; Läufe in eigenem Thread-Pool; Abgleich nach Änderungen. +- Protokoll: pro Lauf eine Zeile (auch übersprungene Termine), Einzeldateien + in `run_items` (max. 2000 pro Lauf); Regelzähler letzter Lauf, letzter Erfolg, + Fehler in Folge. +- Aufräumen der Protokolle nach konfigurierbarer Frist (Standard 30 Tage). +- Gesundheitsprüfung: Scheduler-Herzschlag, Anbindungen im Fehlerzustand, + Regeln ohne Erfolg seit mehr als dem Dreifachen ihres Intervalls. +- Webhook-Benachrichtigung mit frei wählbarer Vorlage (JSON-sicher), Modus + „bei jedem Fehler“ oder „nach N Fehlern in Folge“, optional Entwarnung; + auch für Anbindungen, deren Wiederherstellung scheitert. +- Dienststart: Migration → Schlüssel → Anbindungen wiederherstellen → Scheduler; + beim Beenden werden Mounts nicht ausgehängt. ### Behoben - Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den diff --git a/DEVLOG.md b/DEVLOG.md index 760504c..7a5ca7e 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -10,10 +10,37 @@ Zeiten in Ortszeit (Europe/Berlin). | 2026-10-10 | 17:21 | 17:27 | 0:06 | 1 | Gerüst, Datenmodell, Migrationen | | 2026-10-10 | 18:35 | 18:48 | 0:13 | 2 | Mount-Schicht, Validierung, SFTP | | 2026-10-10 | 18:53 | 19:07 | 0:14 | 3 | Mover-Logik | -| **Summe** | | | **0:33** | | | +| 2026-10-10 | 19:10 | 19:17 | 0:07 | 4 | Scheduler, Protokoll, Benachrichtigung | +| **Summe** | | | **0:40** | | | --- +## 2026-10-10 · 19:10–19:17 · Schritt 4: Scheduler und Protokollierung + +**Erledigt** +- `settings_store.py` (Laufzeit-Einstellungen), `notify.py` (Webhook), + `runs.py` (Lauf ausführen + protokollieren, Zähler, Aufräumen), + `health.py`, `scheduler.py`, `app_state.py` (Verdrahtung, Start/Stop). +- 24 neue Tests (207 gesamt). Mutationsprobe: 6/6 erkannt. + +**Entscheidungen** +- Erfolg = `ok` oder `partial` (Werkzeug arbeitet); Fehlschlag = `error` oder + `partial` (für Benachrichtigungen). Übersprungen zählt nicht; manueller + Probelauf ändert keine Zähler. +- Bezug für „zu lange kein Erfolg“: spätester von letztem Erfolg, Anlage der + Regel und Dienststart – sonst wäre nach einem Neustart sofort alles rot. +- „Nach N Fehlern“ meldet genau einmal beim Erreichen der Schwelle, danach + Entwarnung beim ersten Erfolg (abschaltbar). +- `max_instances=3` im Scheduler, damit überlappende Termine beim eigenen + Regel-Lock ankommen und protokolliert werden, statt dass APScheduler sie still verwirft. +- Leere Läufe (alle 30 s) loggen nur auf DEBUG; in `runs` stehen sie trotzdem + (die Oberfläche filtert standardmäßig auf Läufe mit Aktivität). +- Webhook-URL wird nie geloggt (enthält oft ein Token). + +**Stolpersteine** +- Zwei Testfehler lagen in den Tests selbst (Fake-Symlink bei `resolve()`, + Jitter in der Zeitschranke vergessen). + ## 2026-10-10 · 18:53–19:07 · Schritt 3: Mover-Logik **Erledigt** diff --git a/README.md b/README.md index 4290b73..7ec2d39 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ Selbst gehosteter Dienst mit Web-Oberfläche, der SMB/CIFS- und NFS-Freigaben verwaltet und Dateien zeitgesteuert zwischen ihnen verschiebt. -> Status: in Entwicklung (Schritt 3 von 6). Vollständige Doku folgt mit Schritt 6. +> Status: in Entwicklung (Schritt 4 von 6). Vollständige Doku folgt mit Schritt 6. ## Entwicklung diff --git a/src/netmover/app_state.py b/src/netmover/app_state.py new file mode 100644 index 0000000..2693ffe --- /dev/null +++ b/src/netmover/app_state.py @@ -0,0 +1,83 @@ +"""Verdrahtung aller Bausteine – einmal pro Prozess. + +Start: Datenbank migrieren → Schlüssel laden → Anbindungen mit ``auto_mount`` +wiederherstellen (Ergebnis protokollieren) → erst dann Scheduler starten. +Ende: Scheduler stoppen, Mounts **nicht** aushängen (andere könnten sie nutzen). +""" + +from __future__ import annotations + +from datetime import datetime + +from netmover.config import Settings +from netmover.crypto import SecretBox, load_or_create_key +from netmover.db import make_engine, make_session_factory, upgrade +from netmover.health import Health, check_health +from netmover.logging_setup import event, get_logger +from netmover.models import utcnow +from netmover.mounting.backend import MountBackend +from netmover.mounting.fake import FakeMountBackend +from netmover.mounting.linux import LinuxMountBackend +from netmover.mounting.manager import MountManager, RestoreResult +from netmover.mover.runner import RuleRunner +from netmover.notify import Notifier +from netmover.runs import RunService +from netmover.scheduler import RuleScheduler + +log = get_logger(__name__) + + +def make_backend(settings: Settings) -> MountBackend: + if settings.mount_backend == "fake": + return FakeMountBackend(settings.data_dir / "fake") + return LinuxMountBackend(settings.run_dir) + + +class NetMover: + def __init__( + self, + settings: Settings, + backend: MountBackend | None = None, + notifier: Notifier | None = None, + ) -> None: + self.settings = settings + settings.data_dir.mkdir(parents=True, exist_ok=True) + url = settings.effective_database_url + upgrade(url) + self.db_engine = make_engine(url) + self.session_factory = make_session_factory(self.db_engine) + self.secrets = SecretBox(load_or_create_key(settings.secret_key_path)) + self.backend = backend or make_backend(settings) + self.notifier = notifier or Notifier(self.session_factory) + self.mounts = MountManager( + self.session_factory, + self.backend, + self.secrets, + settings.mount_base, + notifier=self.notifier.mount_event, + ) + self.runner = RuleRunner(self.session_factory, self.mounts) + self.runs = RunService(self.session_factory, self.runner, self.notifier) + self.scheduler = RuleScheduler(self.session_factory, self.runs) + self.started_at: datetime = utcnow() + self.restore_results: list[RestoreResult] = [] + + def start(self) -> None: + """Muss im laufenden Event-Loop aufgerufen werden (AsyncIOScheduler).""" + self.started_at = utcnow() + self.restore_results = self.mounts.restore_all() + ok = sum(r.ok for r in self.restore_results) + event(log, "startup.mounts_restored", ok=ok, failed=len(self.restore_results) - ok) + self.scheduler.start() + + def stop(self) -> None: + self.scheduler.shutdown() + self.db_engine.dispose() + + def health(self) -> Health: + return check_health( + self.session_factory, + scheduler_running=self.scheduler.running, + last_heartbeat=self.scheduler.last_heartbeat, + service_started=self.started_at, + ) diff --git a/src/netmover/health.py b/src/netmover/health.py new file mode 100644 index 0000000..69d2b15 --- /dev/null +++ b/src/netmover/health.py @@ -0,0 +1,75 @@ +"""Gesundheitszustand für ``/healthz`` und die Statusseite (Fallstrick 9). + +Ein Werkzeug, das nichts mehr tut, darf nicht wie eines aussehen, das nichts zu +tun hat. Deshalb: + +* der Scheduler muss leben (Herzschlag nicht älter als 3 Minuten), +* keine aktive Anbindung im Fehlerzustand, +* keine aktive Regel, deren letzter erfolgreicher Lauf länger als das Dreifache + ihres Intervalls zurückliegt. Bezugspunkt ist der späteste von letztem Erfolg, + Anlage der Regel und Dienststart – sonst wäre direkt nach einem Neustart alles rot. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timedelta + +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from netmover.db import session_scope +from netmover.models import Mount, MountState, Rule, utcnow + +STALE_FACTOR = 3 +HEARTBEAT_MAX_AGE = timedelta(minutes=3) + + +@dataclass +class Health: + ok: bool = True + problems: list[str] = field(default_factory=list) + stale_rules: set[int] = field(default_factory=set) + + def fail(self, problem: str) -> None: + self.ok = False + self.problems.append(problem) + + +def rule_is_stale(rule: Rule, service_started: datetime, now: datetime) -> bool: + if not rule.enabled: + return False + reference = max( + d for d in (rule.last_success_at, rule.created_at, service_started) if d is not None + ) + return now - reference > timedelta(seconds=STALE_FACTOR * rule.interval_seconds) + + +def check_health( + session_factory: sessionmaker[Session], + *, + scheduler_running: bool, + last_heartbeat: datetime | None, + service_started: datetime, + now: datetime | None = None, +) -> Health: + now = now or utcnow() + health = Health() + if not scheduler_running: + health.fail("Scheduler läuft nicht") + elif last_heartbeat is None or now - last_heartbeat > HEARTBEAT_MAX_AGE: + health.fail("Scheduler reagiert nicht (kein Herzschlag)") + with session_scope(session_factory) as s: + for m in s.scalars(select(Mount).where(Mount.enabled.is_(True))): + if m.state is MountState.ERROR: + health.fail(f"Anbindung „{m.name}“ im Fehlerzustand: {m.last_error or ''}".strip()) + for r in s.scalars(select(Rule).where(Rule.enabled.is_(True))): + if rule_is_stale(r, service_started, now): + health.stale_rules.add(r.id) + ago = ( + "noch nie" + if r.last_success_at is None + else f"seit {r.last_success_at:%Y-%m-%d %H:%M} UTC" + ) + health.fail(f"Regel „{r.name}“: kein erfolgreicher Lauf {ago}") + return health diff --git a/src/netmover/runs.py b/src/netmover/runs.py new file mode 100644 index 0000000..7cc4d66 --- /dev/null +++ b/src/netmover/runs.py @@ -0,0 +1,171 @@ +"""Läufe ausführen und protokollieren. + +Pro Lauf eine Zeile in ``runs`` (auch für übersprungene Termine, Fallstrick 8), +die Einzeldateien in ``run_items``. Danach die Regelzähler: + +* ``last_run_at`` – jeder Lauf außer übersprungenen, +* ``last_success_at`` – Status ``ok`` oder ``partial`` (das Werkzeug hat gearbeitet), +* ``consecutive_failures`` – ``error`` und ``partial`` zählen hoch, ``ok`` setzt zurück. + +Ein manuell gestarteter Probelauf verändert keine Zähler. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta + +from sqlalchemy import delete, select +from sqlalchemy.orm import Session, sessionmaker + +from netmover.db import session_scope +from netmover.logging_setup import event, get_logger +from netmover.models import FileFailure, Rule, Run, RunItem, RunStatus, RunTrigger, utcnow +from netmover.mover.engine import RunResult +from netmover.mover.runner import RuleRunner +from netmover.notify import Notifier +from netmover.settings_store import load_settings + +log = get_logger(__name__) + +MAX_ITEMS_PER_RUN = 2000 +SUCCESS = frozenset({RunStatus.OK, RunStatus.PARTIAL}) +FAILURE = frozenset({RunStatus.ERROR, RunStatus.PARTIAL}) + + +class RunService: + def __init__( + self, + session_factory: sessionmaker[Session], + runner: RuleRunner, + notifier: Notifier | None = None, + ) -> None: + self.session_factory = session_factory + self.runner = runner + self.notifier = notifier + + def execute( + self, + rule_id: int, + trigger: RunTrigger = RunTrigger.SCHEDULE, + dry_run: bool | None = None, + ) -> int | None: + """Führt die Regel aus, protokolliert den Lauf und liefert die Run-ID + (None, wenn die Regel inzwischen gelöscht wurde).""" + started = utcnow() + result = self.runner.run(rule_id, dry_run=dry_run) + finished = utcnow() + count_it = not (trigger is RunTrigger.MANUAL and dry_run) + return self.record(rule_id, trigger, result, started, finished, update_counters=count_it) + + def record( + self, + rule_id: int, + trigger: RunTrigger, + result: RunResult, + started: datetime, + finished: datetime, + *, + update_counters: bool = True, + ) -> int | None: + notify: tuple[str, bool, int, str] | None = None + with session_scope(self.session_factory) as s: + rule = s.get(Rule, rule_id) + if rule is None: + return None + message = result.message + items = result.items + if len(items) > MAX_ITEMS_PER_RUN: + extra = len(items) - MAX_ITEMS_PER_RUN + items = items[:MAX_ITEMS_PER_RUN] + note = f"{extra} weitere Dateien nicht einzeln protokolliert" + message = f"{message}; {note}" if message else note + run = Run( + rule_id=rule_id, + started_at=started, + finished_at=finished, + status=result.status, + trigger=trigger, + dry_run=result.dry_run, + seen=result.seen, + moved=result.moved, + skipped=result.skipped, + failed=result.failed, + message=message, + ) + s.add(run) + s.flush() + s.add_all( + RunItem( + run_id=run.id, + src=i.src, + dst=i.dst, + status=i.status, + size=i.size, + message=i.message, + ) + for i in items + ) + if update_counters and result.status is not RunStatus.SKIPPED: + previous = rule.consecutive_failures + rule.last_run_at = finished + if result.status in SUCCESS: + rule.last_success_at = finished + if result.status in FAILURE: + rule.consecutive_failures += 1 + notify = (rule.name, True, rule.consecutive_failures, self._summary(result)) + else: + rule.consecutive_failures = 0 + if previous: + notify = (rule.name, False, previous, "") + run_id = run.id + rule_name = rule.name + + level = 20 if result.status in (RunStatus.OK, RunStatus.SKIPPED) else 30 + if result.status is RunStatus.OK and not result.items: + level = 10 # leere Läufe alle 30 s nur auf DEBUG + event( + log, + "run.finished", + level=level, + rule=rule_name, + run=run_id, + status=result.status.value, + trigger=trigger.value, + dry_run=result.dry_run, + seen=result.seen, + moved=result.moved, + skipped=result.skipped, + failed=result.failed, + took_ms=int((finished - started).total_seconds() * 1000), + message=result.message or "", + ) + if notify is not None and self.notifier is not None: + self.notifier.rule_result(*notify) + return run_id + + @staticmethod + def _summary(result: RunResult) -> str: + if result.status is RunStatus.ERROR: + return result.message or "Lauf fehlgeschlagen" + failed = [i for i in result.items if i.status.value == "failed"][:5] + lines = [f"{result.failed} Datei(en) fehlgeschlagen:"] + lines += [f"• {i.src}: {i.message}" for i in failed] + return "\n".join(lines) + + +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.""" + days = load_settings(session_factory).retention_days + cutoff = (now or utcnow()) - timedelta(days=days) + with session_scope(session_factory) as s: + old_ids = list(s.scalars(select(Run.id).where(Run.started_at < cutoff))) + for i in range(0, len(old_ids), 500): + chunk = old_ids[i : i + 500] + s.execute(delete(RunItem).where(RunItem.run_id.in_(chunk))) + s.execute(delete(Run).where(Run.id.in_(chunk))) + failures = s.execute(delete(FileFailure).where(FileFailure.last_attempt_at < cutoff)) + removed_failures = getattr(failures, "rowcount", 0) or 0 + stats = {"runs": len(old_ids), "file_failures": removed_failures} + event(log, "cleanup.done", retention_days=days, **stats) + return stats diff --git a/src/netmover/scheduler.py b/src/netmover/scheduler.py new file mode 100644 index 0000000..940d6e3 --- /dev/null +++ b/src/netmover/scheduler.py @@ -0,0 +1,146 @@ +"""Zeitsteuerung mit APScheduler (``AsyncIOScheduler``). + +* Ein Job pro aktiver Regel im jeweiligen Intervall. Erster Start mit kleinem + zufälligem Versatz, dazu Jitter pro Termin – damit nicht alle Regeln + gleichzeitig starten (Fallstrick 8). +* Die Läufe selbst sind blockierend und laufen in einem eigenen Thread-Pool, + nie im Event-Loop. Überlappung regelt der Lock im :class:`RuleRunner`, der den + übersprungenen Termin auch protokolliert – deshalb ``max_instances`` > 1. +* Herzschlag-Job für ``/healthz`` und stündliches Aufräumen der Protokolle. +""" + +from __future__ import annotations + +import asyncio +import random +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.interval import IntervalTrigger +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from netmover.db import session_scope +from netmover.logging_setup import event, get_logger +from netmover.models import Rule, RunTrigger, utcnow +from netmover.runs import RunService, cleanup + +log = get_logger(__name__) + +MAX_START_OFFSET = 10.0 +HEARTBEAT_SECONDS = 30 +CLEANUP_SECONDS = 3600 + + +def _job_id(rule_id: int) -> str: + return f"rule:{rule_id}" + + +class RuleScheduler: + def __init__( + self, + session_factory: sessionmaker[Session], + service: RunService, + *, + rng: random.Random | None = None, + workers: int = 8, + ) -> None: + self.session_factory = session_factory + self.service = service + self._rng = rng or random.Random() # noqa: S311 – nur Startversatz + self._executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="nm-run") + self.scheduler = AsyncIOScheduler( + timezone=UTC, + job_defaults={"coalesce": True, "max_instances": 3}, + ) + self._intervals: dict[int, int] = {} + self.last_heartbeat: datetime | None = None + + # -- Lebenszyklus --------------------------------------------------------------- + + def start(self) -> None: + self.scheduler.start() + self.heartbeat() + self.scheduler.add_job( + self.heartbeat, IntervalTrigger(seconds=HEARTBEAT_SECONDS), id="heartbeat" + ) + self.scheduler.add_job( + self._cleanup, + IntervalTrigger(seconds=CLEANUP_SECONDS, start_date=utcnow() + timedelta(minutes=1)), + id="cleanup", + ) + self.sync() + event(log, "scheduler.started", rules=len(self._intervals)) + + def shutdown(self) -> None: + if self.scheduler.running: + self.scheduler.shutdown(wait=False) + # Laufende Worker nicht abwarten: abgebrochene Übertragungen hinterlassen + # höchstens .part-Dateien, die Quelle bleibt unangetastet. + self._executor.shutdown(wait=False, cancel_futures=True) + event(log, "scheduler.stopped") + + @property + def running(self) -> bool: + return bool(self.scheduler.running) + + def heartbeat(self) -> None: + self.last_heartbeat = utcnow() + + # -- Regeln ------------------------------------------------------------------------- + + def sync(self) -> None: + """Jobs an die Regeln in der Datenbank angleichen (nach jeder Änderung).""" + with session_scope(self.session_factory) as s: + wanted = { + rid: interval + for rid, interval in s.execute( + select(Rule.id, Rule.interval_seconds).where(Rule.enabled.is_(True)) + ).all() + } + for rid in list(self._intervals): + if rid not in wanted: + if self.scheduler.get_job(_job_id(rid)): + self.scheduler.remove_job(_job_id(rid)) + del self._intervals[rid] + for rid, interval in wanted.items(): + if self._intervals.get(rid) == interval and self.scheduler.get_job(_job_id(rid)): + continue + offset = self._rng.uniform(0, min(float(interval), MAX_START_OFFSET)) + trigger = IntervalTrigger( + seconds=interval, + start_date=utcnow() + timedelta(seconds=offset), + jitter=max(1, min(5, interval // 10)), + ) + self.scheduler.add_job( + self._run_rule, + trigger, + args=[rid], + id=_job_id(rid), + name=f"Regel {rid}", + replace_existing=True, + misfire_grace_time=interval, + ) + self._intervals[rid] = interval + + def next_run(self, rule_id: int) -> datetime | None: + job = self.scheduler.get_job(_job_id(rule_id)) + return job.next_run_time if job is not None else None + + async def _run_rule(self, rule_id: int) -> None: + loop = asyncio.get_running_loop() + await loop.run_in_executor( + self._executor, self.service.execute, rule_id, RunTrigger.SCHEDULE + ) + + async def run_now(self, rule_id: int, dry_run: bool | None = None) -> int | None: + """„Jetzt ausführen“ / „Probelauf“ aus der Oberfläche.""" + loop = asyncio.get_running_loop() + return await loop.run_in_executor( + self._executor, self.service.execute, rule_id, RunTrigger.MANUAL, dry_run + ) + + async def _cleanup(self) -> None: + loop = asyncio.get_running_loop() + await loop.run_in_executor(self._executor, cleanup, self.session_factory) diff --git a/tests/test_runs.py b/tests/test_runs.py new file mode 100644 index 0000000..9fd3576 --- /dev/null +++ b/tests/test_runs.py @@ -0,0 +1,395 @@ +"""Protokollierung, Regelzähler, Benachrichtigung, Aufräumen, Health (Fallstrick 9).""" + +from __future__ import annotations + +import json +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 ( + FileFailure, + ItemStatus, + Mount, + MountState, + Rule, + Run, + RunItem, + RunStatus, + RunTrigger, + utcnow, +) +from netmover.mover.engine import ItemResult, RunResult +from netmover.mover.runner import RuleRunner +from netmover.notify import Notification, Notifier, render +from netmover.runs import RunService, cleanup +from netmover.settings_store import AppSettings, NotifyMode, load_settings, save_settings +from tests.conftest import MountFactory + + +class FakeRunner: + def __init__(self) -> None: + self.results: list[RunResult] = [] + + def run(self, rule_id: int, *, dry_run: bool | None = None) -> RunResult: + 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 + + +@pytest.fixture +def sent() -> list[tuple[str, str, str]]: + return [] + + +@pytest.fixture +def notifier(session_factory: sessionmaker[Session], sent: list[tuple[str, str, str]]) -> Notifier: + return Notifier(session_factory, sender=lambda *a: sent.append(a), background=False) + + +@pytest.fixture +def fake_runner() -> FakeRunner: + return FakeRunner() + + +@pytest.fixture +def service( + session_factory: sessionmaker[Session], fake_runner: FakeRunner, notifier: Notifier +) -> RunService: + return RunService(session_factory, fake_runner, notifier) # type: ignore[arg-type] + + +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 _enable_webhook(sf: sessionmaker[Session], mode: NotifyMode, after: int = 3) -> None: + save_settings( + sf, AppSettings(webhook_url="http://hook.invalid/x", notify_mode=mode, notify_after=after) + ) + + +def ok() -> RunResult: + return RunResult( + RunStatus.OK, seen=1, moved=1, items=[ItemResult("a.pdf", ItemStatus.MOVED, "/x/a.pdf", 3)] + ) + + +def err() -> RunResult: + return RunResult(RunStatus.ERROR, message="„scans“ ist nicht eingehängt") + + +def partial() -> RunResult: + return RunResult( + RunStatus.PARTIAL, + seen=2, + moved=1, + failed=1, + items=[ItemResult("b.pdf", ItemStatus.FAILED, None, 1, "kaputt (Versuch 1)")], + ) + + +def skipped() -> RunResult: + return RunResult(RunStatus.SKIPPED, message="läuft noch") + + +def test_run_and_items_are_recorded( + service: RunService, + fake_runner: FakeRunner, + rule_id: int, + session_factory: sessionmaker[Session], +) -> None: + fake_runner.results.append(ok()) + run_id = service.execute(rule_id) + with session_factory() as s: + run = s.get(Run, run_id) + assert run is not None + assert (run.status, run.seen, run.moved, run.trigger) == ( + RunStatus.OK, + 1, + 1, + RunTrigger.SCHEDULE, + ) + assert run.finished_at is not None and run.finished_at >= run.started_at + items = s.scalars(select(RunItem)).all() + assert [(i.src, i.dst, i.status) for i in items] == [ + ("a.pdf", "/x/a.pdf", ItemStatus.MOVED) + ] + r = _rule(session_factory, rule_id) + assert r.last_success_at is not None and r.consecutive_failures == 0 + + +def test_counters( + service: RunService, + fake_runner: FakeRunner, + rule_id: int, + session_factory: sessionmaker[Session], +) -> None: + fake_runner.results += [err(), partial(), skipped()] + service.execute(rule_id) + r = _rule(session_factory, rule_id) + assert r.consecutive_failures == 1 and r.last_success_at is None and r.last_run_at is not None + service.execute(rule_id) + r = _rule(session_factory, rule_id) + assert r.consecutive_failures == 2 and r.last_success_at is not None # partial: Werkzeug lebt + last_run = r.last_run_at + service.execute(rule_id) # übersprungen: protokolliert, Zähler unverändert + r = _rule(session_factory, rule_id) + assert r.consecutive_failures == 2 and r.last_run_at == last_run + with session_factory() as s: + assert [x.status for x in s.scalars(select(Run).order_by(Run.id))] == [ + RunStatus.ERROR, + RunStatus.PARTIAL, + RunStatus.SKIPPED, + ] + + +def test_manual_dry_run_does_not_touch_counters( + service: RunService, + fake_runner: FakeRunner, + rule_id: int, + session_factory: sessionmaker[Session], +) -> None: + fake_runner.results.append(RunResult(RunStatus.ERROR, dry_run=True, message="x")) + service.execute(rule_id, RunTrigger.MANUAL, dry_run=True) + r = _rule(session_factory, rule_id) + assert r.consecutive_failures == 0 and r.last_run_at is None + + +def test_deleted_rule_is_not_recorded(service: RunService, fake_runner: FakeRunner) -> None: + fake_runner.results.append(err()) + assert service.execute(4711) is None + + +def test_items_are_capped( + service: RunService, + fake_runner: FakeRunner, + rule_id: int, + session_factory: sessionmaker[Session], + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr("netmover.runs.MAX_ITEMS_PER_RUN", 3) + fake_runner.results.append( + RunResult(RunStatus.OK, items=[ItemResult(f"{i}.pdf", ItemStatus.MOVED) for i in range(5)]) + ) + run_id = service.execute(rule_id) + with session_factory() as s: + assert len(s.scalars(select(RunItem)).all()) == 3 + run = s.get(Run, run_id) + assert run is not None and "2 weitere" in (run.message or "") + + +# -- Benachrichtigung ------------------------------------------------------------------------- + + +def test_notify_every_failure( + service: RunService, + fake_runner: FakeRunner, + rule_id: int, + session_factory: sessionmaker[Session], + sent: list[tuple[str, str, str]], +) -> None: + _enable_webhook(session_factory, NotifyMode.EVERY) + fake_runner.results += [err(), err(), ok()] + for _ in range(3): + service.execute(rule_id) + bodies = [json.loads(b) for _, _, b in sent] + assert [b["event"] for b in bodies] == ["rule.failed", "rule.failed", "rule.recovered"] + assert bodies[0]["message"] == "„scans“ ist nicht eingehängt" + assert "2× in Folge" in bodies[1]["title"] + + +def test_notify_after_n_only_once_at_threshold( + service: RunService, + fake_runner: FakeRunner, + rule_id: int, + session_factory: sessionmaker[Session], + sent: list[tuple[str, str, str]], +) -> None: + _enable_webhook(session_factory, NotifyMode.AFTER_N, after=3) + fake_runner.results += [err(), err(), err(), err(), partial(), ok()] + events = [] + for _ in range(6): + service.execute(rule_id) + events.append(len(sent)) + assert events == [0, 0, 1, 1, 1, 2] + assert json.loads(sent[-1][2])["event"] == "rule.recovered" + + +def test_no_recovery_message_below_threshold( + service: RunService, + fake_runner: FakeRunner, + rule_id: int, + session_factory: sessionmaker[Session], + sent: list[tuple[str, str, str]], +) -> None: + _enable_webhook(session_factory, NotifyMode.AFTER_N, after=3) + fake_runner.results += [err(), ok()] + service.execute(rule_id) + service.execute(rule_id) + assert sent == [] + + +def test_notify_off_by_default( + service: RunService, fake_runner: FakeRunner, rule_id: int, sent: list[tuple[str, str, str]] +) -> None: + fake_runner.results.append(err()) + service.execute(rule_id) + assert sent == [] + + +def test_render_json_escapes_values() -> None: + n = Notification("rule.failed", 'Titel mit "Anführung"', "Zeile 1\nZeile 2 \\ $x", "r") + body = render('{"t": "$title", "m": "$message", "x": "$unbekannt"}', "application/json", n) + data = json.loads(body) + assert data == {"t": 'Titel mit "Anführung"', "m": "Zeile 1\nZeile 2 \\ $x", "x": "$unbekannt"} + assert render("$message", "text/plain", n) == "Zeile 1\nZeile 2 \\ $x" + + +def test_sender_failure_is_swallowed(session_factory: sessionmaker[Session]) -> None: + def broken(*a: object) -> None: + raise ConnectionError("weg") + + _enable_webhook(session_factory, NotifyMode.EVERY) + Notifier(session_factory, sender=broken, background=False).send(Notification("x", "t", "m")) + + +def test_settings_roundtrip(session_factory: sessionmaker[Session]) -> None: + assert load_settings(session_factory) == AppSettings() + save_settings(session_factory, AppSettings(retention_days=7, notify_mode=NotifyMode.EVERY)) + loaded = load_settings(session_factory) + assert loaded.retention_days == 7 and loaded.notify_mode is NotifyMode.EVERY + + +# -- Aufräumen --------------------------------------------------------------------------------- + + +def test_cleanup_removes_old_runs_with_items( + service: RunService, + fake_runner: FakeRunner, + rule_id: int, + session_factory: sessionmaker[Session], +) -> None: + fake_runner.results += [ok(), ok()] + old = service.execute(rule_id) + new = service.execute(rule_id) + with session_scope(session_factory) as s: + run = s.get(Run, old) + assert run is not None + run.started_at = utcnow() - timedelta(days=31) + s.add( + FileFailure( + rule_id=rule_id, + src_relpath="x", + size=1, + mtime_ns=1, + attempts=1, + last_attempt_at=utcnow() - timedelta(days=40), + ) + ) + stats = cleanup(session_factory) + assert stats == {"runs": 1, "file_failures": 1} + with session_factory() as s: + assert [r.id for r in s.scalars(select(Run))] == [new] + assert len(s.scalars(select(RunItem)).all()) == 1 + + +# -- Health (Fallstrick 9) ------------------------------------------------------------------- + + +def _health(sf: sessionmaker[Session], started_ago: float = 3600, **kw: object): # type: ignore[no-untyped-def] + now = utcnow() + args: dict[str, object] = { + "scheduler_running": True, + "last_heartbeat": now, + "service_started": now - timedelta(seconds=started_ago), + "now": now, + } + args.update(kw) + return check_health(sf, **args) # type: ignore[arg-type] + + +def _age_rule(sf: sessionmaker[Session], rid: int, seconds: float | None) -> None: + with session_scope(sf) as s: + r = s.get(Rule, rid) + assert r is not None + r.created_at = utcnow() - timedelta(days=1) + r.last_success_at = None if seconds is None else utcnow() - timedelta(seconds=seconds) + + +def test_health_ok(session_factory: sessionmaker[Session], rule_id: int) -> None: + _age_rule(session_factory, rule_id, 60) # 2× Intervall + h = _health(session_factory) + assert h.ok, h.problems + + +def test_health_stale_rule_after_three_intervals( + session_factory: sessionmaker[Session], rule_id: int +) -> None: + _age_rule(session_factory, rule_id, 91) + h = _health(session_factory) + assert not h.ok and h.stale_rules == {rule_id} + assert "kein erfolgreicher Lauf" in h.problems[0] + + +def test_health_never_succeeded_but_service_just_started( + session_factory: sessionmaker[Session], rule_id: int +) -> None: + _age_rule(session_factory, rule_id, None) + assert _health(session_factory, started_ago=30).ok + assert not _health(session_factory, started_ago=120).ok + + +def test_health_disabled_rule_is_ignored( + session_factory: sessionmaker[Session], rule_id: int +) -> None: + _age_rule(session_factory, rule_id, None) + with session_scope(session_factory) as s: + r = s.get(Rule, rule_id) + assert r is not None + r.enabled = False + assert _health(session_factory).ok + + +def test_health_mount_error_only_for_enabled_mounts( + session_factory: sessionmaker[Session], make_mount: MountFactory +) -> None: + mid = make_mount("b", share="b", enabled=False) + with session_scope(session_factory) as s: + m = s.get(Mount, mid) + assert m is not None + m.state = MountState.ERROR + assert _health(session_factory).ok + with session_scope(session_factory) as s: + m = s.get(Mount, mid) + assert m is not None + m.enabled = True + h = _health(session_factory) + assert not h.ok and "Fehlerzustand" in h.problems[0] + + +def test_health_scheduler(session_factory: sessionmaker[Session]) -> None: + assert not _health(session_factory, scheduler_running=False).ok + assert not _health(session_factory, last_heartbeat=utcnow() - timedelta(minutes=5)).ok + + +def test_runner_type_is_compatible(session_factory: sessionmaker[Session], tmp_path: Path) -> None: + # FakeRunner oben muss dieselbe Schnittstelle bieten wie RuleRunner.run + assert RuleRunner.run.__code__.co_kwonlyargcount == 1 diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py new file mode 100644 index 0000000..3395e48 --- /dev/null +++ b/tests/test_scheduler.py @@ -0,0 +1,177 @@ +"""Scheduler: Jobs je Regel, Versatz, Neuabgleich, manuelle Läufe, Dienststart.""" + +from __future__ import annotations + +import asyncio +import random +from collections.abc import Callable, Coroutine +from datetime import timedelta +from pathlib import Path +from typing import Any + +import pytest +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from netmover.app_state import NetMover +from netmover.config import Settings +from netmover.db import session_scope +from netmover.models import Mount, MountState, MountType, Rule, Run, RunStatus, RunTrigger, utcnow +from netmover.mounting.fake import FakeMountBackend +from netmover.mounting.manager import MountManager +from netmover.mover.runner import RuleRunner +from netmover.runs import RunService +from netmover.scheduler import MAX_START_OFFSET, RuleScheduler +from tests.conftest import MountFactory + + +def arun(fn: Callable[[], Coroutine[Any, Any, None]]) -> None: + asyncio.run(fn()) + + +@pytest.fixture +def rules( + make_mount: MountFactory, + session_factory: sessionmaker[Session], + fake_backend: FakeMountBackend, + manager: MountManager, +) -> list[int]: + fake_backend.export("nas", "scans") + mid = make_mount() + manager.mount(mid) + ids = [] + with session_scope(session_factory) as s: + for name, interval, enabled in [("a", 30, True), ("b", 60, True), ("c", 30, False)]: + r = Rule( + name=name, + src_mount_id=mid, + dst_path=str(fake_backend.remote_root), + interval_seconds=interval, + enabled=enabled, + stability_seconds=0, + ) + s.add(r) + s.flush() + ids.append(r.id) + return ids + + +@pytest.fixture +def scheduler(session_factory: sessionmaker[Session], manager: MountManager) -> RuleScheduler: + runner = RuleRunner(session_factory, manager, sleep=lambda _: None) + return RuleScheduler(session_factory, RunService(session_factory, runner), rng=random.Random(1)) # noqa: S311 + + +def test_jobs_follow_rules( + scheduler: RuleScheduler, rules: list[int], session_factory: sessionmaker[Session] +) -> None: + a, b, c = rules + + async def main() -> None: + scheduler.start() + try: + assert scheduler.next_run(a) is not None + assert scheduler.next_run(b) is not None + assert scheduler.next_run(c) is None + # erster Termin mit kleinem Versatz + for rid in (a, b): + nxt = scheduler.next_run(rid) + assert nxt is not None + # Versatz ≤ 10 s plus Jitter ≤ 5 s + assert nxt - utcnow() <= timedelta(seconds=MAX_START_OFFSET + 5 + 1) + + with session_scope(session_factory) as s: + ra, rc = s.get(Rule, a), s.get(Rule, c) + assert ra is not None and rc is not None + ra.enabled = False + rc.enabled = True + rb = s.get(Rule, b) + assert rb is not None + rb.interval_seconds = 120 + scheduler.sync() + assert scheduler.next_run(a) is None + assert scheduler.next_run(c) is not None + job = scheduler.scheduler.get_job(f"rule:{b}") + assert job is not None and job.trigger.interval == timedelta(seconds=120) + finally: + scheduler.shutdown() + + arun(main) + + +def test_run_now_records_manual_run( + scheduler: RuleScheduler, rules: list[int], session_factory: sessionmaker[Session] +) -> None: + a = rules[0] + + async def main() -> None: + scheduler.start() + try: + run_id = await scheduler.run_now(a, dry_run=True) + finally: + scheduler.shutdown() + with session_factory() as s: + run = s.get(Run, run_id) + assert run is not None + assert run.trigger is RunTrigger.MANUAL and run.dry_run and run.status is RunStatus.OK + + arun(main) + + +def test_scheduled_job_runs_in_thread_pool( + scheduler: RuleScheduler, rules: list[int], session_factory: sessionmaker[Session] +) -> None: + a = rules[0] + + async def main() -> None: + scheduler.start() + try: + await scheduler._run_rule(a) + assert scheduler.running and scheduler.last_heartbeat is not None + finally: + scheduler.shutdown() + with session_factory() as s: + runs = s.scalars(select(Run)).all() + assert [(r.rule_id, r.trigger) for r in runs] == [(a, RunTrigger.SCHEDULE)] + + arun(main) + + +def test_app_startup_restores_mounts_before_scheduler(tmp_path: Path) -> None: + settings = Settings( + data_dir=tmp_path / "lib", + run_dir=tmp_path / "run", + mount_base=tmp_path / "mnt", + mount_backend="fake", + ) + backend = FakeMountBackend(tmp_path / "fake") + backend.export("nas", "scans") + app = NetMover(settings, backend=backend) + with session_scope(app.session_factory) as s: + s.add( + Mount( + name="scans", + slug="scans", + type=MountType.CIFS, + server="nas", + share="scans", + mountpoint=str((tmp_path / "mnt" / "scans").resolve()), + ) + ) + assert (tmp_path / "lib" / "secret.key").stat().st_mode & 0o777 == 0o600 + + async def main() -> None: + app.start() + try: + assert [r.ok for r in app.restore_results] == [True] + assert app.scheduler.running + assert app.health().ok + finally: + app.stop() + + arun(main) + # Beim Beenden NICHT aushängen + assert backend.is_mounted((tmp_path / "mnt").resolve() / "scans") + with app.session_factory() as s: + m = s.scalars(select(Mount)).one() + assert m.state is MountState.MOUNTED