HitonabiandClaude Opus 5.5
34360c9644
services/ablaeufe.py: je systemd-Timer der Box der erwartete Takt plus Karenz (LastTriggerUSec über --timestamp=unix), Hermes-Cron-Jobs mit verpasstem next_run_at, der Runner der Gitea-Ampel (Admin-API, Token aus ~/.git-credentials, nie ausgegeben). Überfällig oder gestoppt → Hinweis nach fünf Takten, die Sonntags-Updates rot. GET /api/system/ablaeufe; Panel „Zeitpläne“ unter den Diensten. Tests fragen Gitea nie echt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
90 lines
5.3 KiB
Python
90 lines
5.3 KiB
Python
"""Totmannschalter (services/ablaeufe.py, 25.09.2026): Abläufe, die gar nicht erst laufen, werden ein Hinweis.
|
|
|
|
Die Zeitstempel folgen `systemctl --user show … --timestamp=unix` auf der Box (25.09.2026): „@1790300009“, leer bei
|
|
einem Timer, der noch nie gestartet ist (die Probe-Wiederherstellung läuft erst am ersten Montag im Monat)."""
|
|
|
|
from services import ablaeufe
|
|
|
|
JETZT = 1_790_340_000.0 # 25.09.2026 ~14:40
|
|
BACKUP = next(z for z in ablaeufe.ZEITPLAENE if z.unit == "mc2-backup")
|
|
PROBE = next(z for z in ablaeufe.ZEITPLAENE if z.unit == "mc2-probe-wiederherstellung")
|
|
|
|
|
|
def _zustand(**mehr: str) -> dict[str, str]:
|
|
return {"LoadState": "loaded", "ActiveState": "active", "UnitFileState": "enabled",
|
|
"ActiveEnterTimestamp": "@1790254319", **mehr}
|
|
|
|
|
|
def test_timer_puenktlich_und_ueberfaellig():
|
|
ok = ablaeufe.bewerte_timer(BACKUP, _zustand(LastTriggerUSec="@1790300009", NextElapseUSecRealtime="@1790386222"),
|
|
JETZT)
|
|
assert (ok["status"], ok["letzter"], ok["naechster"]) == ("ok", 1790300009.0, 1790386222.0)
|
|
alt = ablaeufe.bewerte_timer(BACKUP, _zustand(LastTriggerUSec=f"@{int(JETZT - 30 * 3600)}"), JETZT)
|
|
assert alt["status"] == "ueberfaellig" and alt["text"] == "Zuletzt vor 30 Stunden gestartet (täglich 03:30 erwartet)."
|
|
|
|
|
|
def test_nie_gestartet_zaehlt_ab_dem_start_des_timers():
|
|
frisch = ablaeufe.bewerte_timer(PROBE, _zustand(LastTriggerUSec=""), JETZT)
|
|
assert frisch["status"] == "ok" and frisch["letzter"] is None
|
|
lange = ablaeufe.bewerte_timer(PROBE, _zustand(LastTriggerUSec="", ActiveEnterTimestamp=f"@{int(JETZT - 40 * 86400)}"),
|
|
JETZT)
|
|
assert lange["status"] == "ueberfaellig" and lange["text"].startswith("Seit 40 Tagen kein einziger Start")
|
|
|
|
|
|
def test_bewusst_aus_ist_kein_befund_gestoppt_schon():
|
|
aus = ablaeufe.bewerte_timer(BACKUP, _zustand(ActiveState="inactive", UnitFileState="disabled"), JETZT)
|
|
assert aus["status"] == "aus"
|
|
gestoppt = ablaeufe.bewerte_timer(BACKUP, _zustand(ActiveState="inactive"), JETZT)
|
|
assert gestoppt["status"] == "gestoppt" and "startet so nie" in gestoppt["text"]
|
|
assert ablaeufe.bewerte_timer(BACKUP, {}, JETZT) is None # kein systemd (Windows)
|
|
assert ablaeufe.bewerte_timer(BACKUP, {"LoadState": "not-found"}, JETZT) is None
|
|
|
|
|
|
def test_hermes_job_ueberfaellig_wenn_der_planer_steht():
|
|
job = {"id": "195e479e8e30", "name": "Daily News Report", "enabled": True,
|
|
"schedule": {"kind": "cron", "expr": "0 7 * * *", "display": "0 7 * * *"},
|
|
"next_run_at": "2026-09-25T07:00:00+02:00", "last_run_at": "2026-09-24T07:02:05+02:00"}
|
|
e = ablaeufe.bewerte_job(job, JETZT)
|
|
assert e["status"] == "ueberfaellig" and "Planer von Hermes" in e["text"] and e["takt"] == "Hermes-Cron 0 7 * * *"
|
|
assert ablaeufe.bewerte_job({**job, "next_run_at": "2026-09-26T07:00:00+02:00"}, JETZT)["status"] == "ok"
|
|
assert ablaeufe.bewerte_job({**job, "enabled": False}, JETZT) is None
|
|
|
|
|
|
def test_gitea_runner():
|
|
assert ablaeufe.bewerte_runner(None) == [] # kein Token oder Gitea weg: still
|
|
assert ablaeufe.bewerte_runner([])[0]["status"] == "gestoppt"
|
|
(online,) = ablaeufe.bewerte_runner([{"id": 1, "name": "arcane-runner", "status": "online", "disabled": False}])
|
|
assert online["status"] == "ok"
|
|
(weg,) = ablaeufe.bewerte_runner([{"id": 1, "name": "arcane-runner", "status": "offline"}])
|
|
assert weg["status"] == "gestoppt" and "offline" in weg["text"]
|
|
|
|
|
|
def test_gitea_token_aus_den_git_zugaengen(tmp_path, monkeypatch):
|
|
datei = tmp_path / ".git-credentials"
|
|
datei.write_text("https://oauth2:anderes@github.com\nhttp://Hitonabi:geheim123@192.168.178.153:3000\n",
|
|
encoding="utf-8")
|
|
monkeypatch.setattr(ablaeufe, "GIT_ZUGANG", datei)
|
|
monkeypatch.setattr(ablaeufe, "GITEA_URL", "http://192.168.178.153:3000")
|
|
assert ablaeufe._gitea_token() == "geheim123"
|
|
monkeypatch.setattr(ablaeufe, "GIT_ZUGANG", tmp_path / "fehlt")
|
|
assert ablaeufe._gitea_token() is None
|
|
|
|
|
|
def test_befunde_rot_nur_fuer_die_sonntags_updates(monkeypatch):
|
|
liste = [
|
|
{"id": "timer:mc2-autoupdate", "name": "Updates am Sonntag", "art": "zeitplan", "status": "ueberfaellig",
|
|
"text": "Zuletzt vor 9 Tagen gestartet."},
|
|
{"id": "timer:mc2-backup", "name": "Sicherung der Box", "art": "zeitplan", "status": "gestoppt", "text": "…"},
|
|
{"id": "gitea:runner:1", "name": "Gitea-Ampel (arcane-runner)", "art": "ampel", "status": "gestoppt", "text": "…"},
|
|
{"id": "timer:mc2-radar", "name": "Modell-Radar", "art": "zeitplan", "status": "aus", "text": None},
|
|
{"id": "hermes:1", "name": "Daily News Report", "art": "hermes", "status": "ok", "text": None},
|
|
]
|
|
monkeypatch.setattr(ablaeufe, "ablaeufe", lambda jetzt=None: liste)
|
|
befunde = ablaeufe.pruefe_ueberfaellig()
|
|
assert [(b.id, b.stufe, b.titel) for b in befunde] == [
|
|
("ablauf:timer:mc2-autoupdate", "rot", "Updates am Sonntag ist überfällig"),
|
|
("ablauf:timer:mc2-backup", "gelb", "Sicherung der Box läuft nicht"),
|
|
("ablauf:gitea:runner:1", "gelb", "Gitea-Ampel (arcane-runner) läuft nicht")]
|
|
assert befunde[0].aktionen == [{"id": "protokoll", "label": "Protokoll", "dienst": "mc2-autoupdate"}]
|
|
assert befunde[2].aktionen == [] and all(b.nach_takten == 5 for b in befunde)
|