14e766173c
- POST /api/sources/health/run (202, bei laufendem Run 409) stoesst den Lauf als BackgroundTask an: eigene DB-Connection, run_health_checks + generate_suggestions aus den geteilten Diensten, exakt nach dem Vorbild des Monitor-Nachtjobs daily_source_health_check. Kein Code-Duplikat. - GET /api/sources/health/run-status liefert den In-Memory-Status (health_run.py, pure Funktionen mit Doppelstart-Sperre, unit-getestet). Nach einem Neustart ist der Status idle, ein partieller Stand wird vom naechsten Lauf archiviert und ueberschrieben (wie beim Nachtjob). - Frontend: Knopf Jetzt pruefen + Zuletzt geprueft in der Stats-Leiste der Quellenliste, Polling alle 4 s, Abschluss-Toast mit geprueft/ Probleme/neue Vorschlaege, danach Liste + Aufgaben-Badge neu laden. Beim Oeffnen des Reiters wird ein laufender Run aus anderer Session erkannt und weiterverfolgt. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
51 Zeilen
1.5 KiB
Python
51 Zeilen
1.5 KiB
Python
"""Tests für src/health_run.py - Zustandsübergänge des manuellen Health-Runs."""
|
|
from health_run import new_state, try_start, mark_finished, mark_failed
|
|
|
|
|
|
def test_initial_state_is_idle():
|
|
s = new_state()
|
|
assert s["running"] is False
|
|
assert s["started_at"] is None
|
|
assert s["finished_at"] is None
|
|
assert s["result"] is None
|
|
assert s["error"] is None
|
|
|
|
|
|
def test_try_start_sets_running_and_resets():
|
|
s = new_state()
|
|
s["result"] = {"checked": 1}
|
|
s["error"] = "alt"
|
|
assert try_start(s, now="2026-07-25T12:00:00") is True
|
|
assert s["running"] is True
|
|
assert s["started_at"] == "2026-07-25T12:00:00"
|
|
assert s["finished_at"] is None
|
|
assert s["result"] is None
|
|
assert s["error"] is None
|
|
|
|
|
|
def test_double_start_blocked():
|
|
s = new_state()
|
|
assert try_start(s) is True
|
|
assert try_start(s) is False
|
|
|
|
|
|
def test_mark_finished_allows_restart():
|
|
s = new_state()
|
|
try_start(s, now="2026-07-25T12:00:00")
|
|
mark_finished(s, {"checked": 430, "issues": 110, "suggestions": 3}, now="2026-07-25T12:05:00")
|
|
assert s["running"] is False
|
|
assert s["finished_at"] == "2026-07-25T12:05:00"
|
|
assert s["result"]["checked"] == 430
|
|
assert s["error"] is None
|
|
assert try_start(s) is True
|
|
|
|
|
|
def test_mark_failed_sets_error():
|
|
s = new_state()
|
|
try_start(s)
|
|
mark_failed(s, "Verbindung weg", now="2026-07-25T12:01:00")
|
|
assert s["running"] is False
|
|
assert s["error"] == "Verbindung weg"
|
|
assert s["result"] is None
|
|
assert try_start(s) is True
|