"""Monatliche Probe-Wiederherstellung (services/probe_wiederherstellung.py, 24.09.2026) und ihr Wächter-Hinweis. Die Tests bauen Sicherungen so, wie deploy/backup.sh sie schreibt (Mitglieder „./hermes/…“, absolute Verknüpfungen unter systemd/user), mit einem „lebenden“ Hermes-Ordner daneben — alles im Temp-Ordner, ohne Box und ohne Telegram.""" import io import json import os import tarfile import time from datetime import datetime, timedelta from pathlib import Path import pytest from services import probe_wiederherstellung as pw from services import waechter GEDAECHTNIS = {"MEMORY.md": b"Tobi mag kurze Antworten.\n", "USER.md": b"Name: Tobi\n"} def _lebend(wurzel: Path) -> Path: """Der lebende Stand: Hermes-Ordner und Datenordner.""" hermes = wurzel / "live" / "hermes" for name, inhalt in GEDAECHTNIS.items(): (hermes / "memories").mkdir(parents=True, exist_ok=True) (hermes / "memories" / name).write_bytes(inhalt) (hermes / "memories" / "USER.md.lock").write_bytes(b"") (hermes / "SOUL.md").write_bytes(b"Du bist Lucy.\n") (hermes / "skills" / "news").mkdir(parents=True) (hermes / "skills" / "news" / "SKILL.md").write_bytes(b"# News\n") (hermes / "scripts").mkdir() (hermes / "scripts" / "stack-ist.sh").write_bytes(b"#!/bin/bash\n") (wurzel / "daten").mkdir() (wurzel / "daten" / "mc2-waechter.json").write_text('{"hinweise": {}}', encoding="utf-8") return hermes def _inhalt(hermes: Path, wurzel: Path) -> dict[str, bytes]: """Was backup.sh aus dem lebenden Stand einpacken würde.""" inhalt = {f"hermes/memories/{n}": b for n, b in GEDAECHTNIS.items()} inhalt.update({ "hermes/memories/USER.md.lock": b"", "hermes/SOUL.md": (hermes / "SOUL.md").read_bytes(), "hermes/skills/news/SKILL.md": b"# News\n", "hermes/scripts/stack-ist.sh": b"#!/bin/bash\n", "hermes/cron/jobs.json": json.dumps({"jobs": [{"id": "a", "name": "Daily News Report"}]}).encode(), "hermes/config.yaml": b"model:\n default: hermes\n", "hermes/.env": b"TELEGRAM_BOT_TOKEN=geheim\n", "hermes/desktop-gateway-token": b"geheim\n", "llama-swap/config.yaml": b"models:\n Qwen3.6-35B-A3B:\n cmd: llama-server\n", "box-wart/mc2-waechter.json": (wurzel / "daten" / "mc2-waechter.json").read_bytes(), "systemd/user/mission-control-2.service": b"[Service]\n", "MANIFEST.txt": b"mc2-state backup\n", }) return inhalt def _sicherung(ordner: Path, inhalt: dict[str, bytes | None], beginn: datetime | None = None) -> Path: """Ein Tarball wie von backup.sh: Name mit Startzeit, Mitglieder „./…“, dazu eine absolute Verknüpfung.""" beginn = beginn or datetime.now(pw.LOCAL_TZ) + timedelta(minutes=10) # „nach“ dem lebenden Stand ordner.mkdir(parents=True, exist_ok=True) tarball = ordner / f"mc2-state-{beginn:%Y%m%d-%H%M%S}.tar.gz" with tarfile.open(tarball, "w:gz") as tar: for rel, daten in inhalt.items(): if daten is None: continue info = tarfile.TarInfo(f"./{rel}") info.size, info.mode, info.mtime = len(daten), 0o600, time.time() tar.addfile(info, io.BytesIO(daten)) link = tarfile.TarInfo("./systemd/user/default.target.wants/mission-control-2.service") link.type, link.linkname = tarfile.SYMTYPE, "/home/hitonabi/.config/systemd/user/mission-control-2.service" tar.addfile(link) return tarball @pytest.fixture def box(tmp_path, monkeypatch): hermes = _lebend(tmp_path) monkeypatch.setattr(pw, "SICHERUNGEN_DIR", tmp_path / "backups") monkeypatch.setattr(pw, "ZUSTAND_PATH", tmp_path / "daten" / "mc2-probe-wiederherstellung.json") monkeypatch.setattr(pw, "HERMES_LIVE", hermes) monkeypatch.setattr(pw, "DATEN_LIVE", tmp_path / "daten") monkeypatch.setattr(pw, "LLAMA_SWAP_DROPINS", tmp_path / "gibt-es-hier-nicht") (tmp_path / "tmp").mkdir() monkeypatch.setenv("MC_PROBE_TMP", str(tmp_path / "tmp")) meldungen: list[dict] = [] monkeypatch.setattr(pw, "melden", meldungen.append) return tmp_path, hermes, meldungen def test_gruene_probe_packt_aus_prueft_und_raeumt_auf(box, monkeypatch): tmp, hermes, meldungen = box _sicherung(tmp / "backups", _inhalt(hermes, tmp)) gesehen: dict = {} echt = pw._pruefe_inhalt def spion(wurzel, *args): gesehen.update(wurzel=wurzel, env=(wurzel / "hermes" / ".env").exists(), token=(wurzel / "hermes" / "desktop-gateway-token").exists(), seele=(wurzel / "hermes" / "SOUL.md").is_file()) return echt(wurzel, *args) monkeypatch.setattr(pw, "_pruefe_inhalt", spion) stand = pw.probe() assert stand["ergebnis"] == "gruen", stand["fehler"] assert meldungen == [] assert gesehen["seele"] and not gesehen["env"] and not gesehen["token"] # Geheimnisse bleiben im Archiv assert not gesehen["wurzel"].exists() and list((tmp / "tmp").iterdir()) == [] # Tmp-Ordner ist weg assert any(z.startswith("Gedächtnis: 2 Dateien") for z in stand["geprueft"]) assert "Zugangsdaten (.env) enthalten (nicht ausgepackt)" in stand["geprueft"] assert pw.lese_stand()["ergebnis"] == "gruen" def test_fehlendes_gedaechtnis_ist_rot_und_wird_gemeldet(box): tmp, hermes, meldungen = box inhalt = {k: v for k, v in _inhalt(hermes, tmp).items() if not k.startswith("hermes/memories/")} _sicherung(tmp / "backups", inhalt) stand = pw.probe() assert stand["ergebnis"] == "rot" assert any("Lucys Gedächtnis (hermes/memories) fehlt" in f for f in stand["fehler"]) # Abgleich mit dem lebenden Stand: eine Zeile je Bereich, auch bei vielen Dateien. assert "Gedächtnis: 2 Dateien fehlen in der Sicherung (MEMORY.md, USER.md)." in stand["fehler"] assert len(meldungen) == 1 and pw.lese_stand()["ergebnis"] == "rot" def test_viele_fehlende_dateien_ergeben_eine_zeile(box): tmp, hermes, _ = box for i in range(40): (hermes / "skills" / f"s{i:02d}").mkdir() (hermes / "skills" / f"s{i:02d}" / "SKILL.md").write_bytes(b"# x\n") _sicherung(tmp / "backups", _inhalt(hermes, tmp)) fehler = pw.probe()["fehler"] assert fehler == ["Skills: 40 Dateien fehlen in der Sicherung (s00/SKILL.md, s01/SKILL.md, s02/SKILL.md …)."] def test_abweichendes_gedaechtnis_faellt_auf(box): """Stand eine Datei schon vor der Sicherung so da, muss die Sicherung sie unverändert enthalten.""" tmp, hermes, _ = box inhalt = _inhalt(hermes, tmp) inhalt["hermes/memories/USER.md"] = b"Name: jemand anderes\n" _sicherung(tmp / "backups", inhalt) fehler = pw.probe()["fehler"] assert fehler == ["Gedächtnis: USER.md steht in der Sicherung anders als auf der Box."] def test_nach_der_sicherung_geaenderte_dateien_zaehlen_nicht(box): """Die Sicherung begann vor dem lebenden Stand: Abweichungen sind dann neuere Erinnerungen, kein Fehler.""" tmp, hermes, _ = box inhalt = _inhalt(hermes, tmp) inhalt["hermes/memories/USER.md"] = b"alter Stand\n" _sicherung(tmp / "backups", inhalt, beginn=datetime.now(pw.LOCAL_TZ) - timedelta(hours=2)) assert pw.probe()["ergebnis"] == "gruen" @pytest.mark.parametrize("weg, text", [ ("hermes/SOUL.md", "SOUL.md fehlt in der Sicherung"), ("hermes/skills/news/SKILL.md", "Keine Skills"), ("hermes/scripts/stack-ist.sh", "Keine Cron-Skripte"), ("hermes/config.yaml", "Die Hermes-Config (config.yaml) fehlt"), ("hermes/.env", "Die Zugangsdaten (hermes/.env) fehlen"), ("box-wart/mc2-waechter.json", "Der Box-Wart-Zustand (box-wart/mc2-*.json) fehlt"), ("systemd/user/mission-control-2.service", "Die systemd-Units"), ("llama-swap/config.yaml", "Die llama-swap-Config fehlt"), ("hermes/cron/jobs.json", "Die Hermes-Cron-Jobs"), ]) def test_jeder_pflichtinhalt_zaehlt(box, weg, text): tmp, hermes, _ = box inhalt = _inhalt(hermes, tmp) inhalt[weg] = None _sicherung(tmp / "backups", inhalt) stand = pw.probe() assert stand["ergebnis"] == "rot" assert any(text in f for f in stand["fehler"]), stand["fehler"] def test_unlesbarer_zustand_und_alte_sicherung(box): tmp, hermes, _ = box inhalt = _inhalt(hermes, tmp) inhalt["box-wart/mc2-radar.json"] = b"{kaputt" tarball = _sicherung(tmp / "backups", inhalt) alt = time.time() - 3 * 86400 os.utime(tarball, (alt, alt)) fehler = pw.probe()["fehler"] assert any("Die jüngste Sicherung ist 3 Tage alt" in f for f in fehler) assert "Box-Wart-Zustand unlesbar: mc2-radar.json" in fehler def test_kaputte_und_fehlende_sicherung(box): tmp, hermes, meldungen = box assert pw.probe()["fehler"] == [f"Keine Sicherung gefunden ({tmp / 'backups'})."] tarball = _sicherung(tmp / "backups", _inhalt(hermes, tmp)) tarball.write_bytes(tarball.read_bytes()[:200]) # abgeschnitten wie nach einer vollen Platte stand = pw.probe() assert stand["ergebnis"] == "rot" and stand["fehler"][0].startswith("Die Sicherung lässt sich nicht auspacken") assert list((tmp / "tmp").iterdir()) == [] and len(meldungen) == 2 def test_meldung_geht_in_normaler_dringlichkeit_raus(monkeypatch): """Rot → notify.sh mit Betreff [Sicherung], ohne -d: nachts sammelt notify.sh sie für die Morgenmeldung.""" aufrufe: list[list[str]] = [] monkeypatch.setattr(pw, "_bash", lambda: "bash") monkeypatch.setattr(pw.subprocess, "run", lambda befehl, **kw: aufrufe.append(befehl)) pw.melden({"sicherung": "mc2-state-x.tar.gz", "fehler": ["eins", "zwei", "drei", "vier"]}) befehl = aufrufe[0] assert befehl[:2] == ["bash", str(pw.NOTIFY_SH)] and befehl[2:4] == ["-s", "[Sicherung]"] assert "-d" not in befehl and "Alarm" not in befehl[3] and not befehl[4].startswith("KRITISCH") assert "eins zwei drei (und 1 weitere)" in befehl[4] # --- Wächter ------------------------------------------------------------------------------------------ def _stand(ergebnis: str, tage: float, fehler: list[str] | None = None) -> dict: return {"zeit": time.time() - tage * 86400, "ergebnis": ergebnis, "sicherung": "mc2-state-x.tar.gz", "fehler": fehler or []} def test_waechter_zeigt_gelb_bei_rot_oder_zu_alt(monkeypatch): stand: dict = {} monkeypatch.setattr(waechter.probe_wiederherstellung, "lese_stand", lambda: stand.get("x")) monkeypatch.setattr(waechter, "_systemctl_show", lambda name, system: {"LoadState": "loaded"}) stand["x"] = _stand("gruen", 3) assert waechter.pruefe_probe_wiederherstellung() == [] stand["x"] = _stand("rot", 3, ["Lucys Gedächtnis (hermes/memories) fehlt in der Sicherung oder ist leer.", "b"]) (b,) = waechter.pruefe_probe_wiederherstellung() assert (b.id, b.stufe, b.sofort) == ("probe:wiederherstellung", "gelb", True) assert "Lucys Gedächtnis" in b.text and "(und 1 weitere)" in b.text assert {a["id"] for a in b.aktionen} == {"neustart", "protokoll"} assert all(a["dienst"] == "mc2-probe-wiederherstellung" for a in b.aktionen) stand["x"] = _stand("gruen", 41) (b,) = waechter.pruefe_probe_wiederherstellung() assert b.stufe == "gelb" and "seit 41 Tagen" in b.titel def test_waechter_vor_der_ersten_probe(monkeypatch): """Noch nie gelaufen: kein Hinweis, solange der Timer fehlt (PC, frische Box) — danach gelb mit „Jetzt prüfen“.""" monkeypatch.setattr(waechter.probe_wiederherstellung, "lese_stand", lambda: None) monkeypatch.setattr(waechter, "_systemctl_show", lambda name, system: {"LoadState": "not-found"}) assert waechter.pruefe_probe_wiederherstellung() == [] monkeypatch.setattr(waechter, "_systemctl_show", lambda name, system: {"LoadState": "loaded"}) (b,) = waechter.pruefe_probe_wiederherstellung() assert b.stufe == "gelb" and b.aktionen[0]["label"] == "Jetzt prüfen" def test_knoepfe_des_hinweises_sind_erlaubt(): """„Jetzt prüfen“ und „Protokoll“ laufen über die Allowlist in services.maintenance.""" from services import maintenance assert {"mc2-probe-wiederherstellung", "pbs-backup"} <= maintenance.USER_SERVICES assert "pbs-backup" in waechter.TIMER_DIENSTE and waechter.TIMER_DIENSTE["pbs-backup"][1] is False assert waechter.pruefe_probe_wiederherstellung in waechter.PRUEFUNGEN