"""Protokoll des Homelab-Teils (services/homelab/protokoll.py, 24.09.2026): alle Quellen, sprechende deutsche Titel, neueste zuerst — und keine Geheimnisse. Dazu die längere Aufbewahrung im Kanal und die Ereignisse der Pflege.""" import json from collections import Counter from datetime import datetime from pathlib import Path import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from kern import einstellungen as einstellungen_mod from kern.zeit import LOCAL_TZ from services import waechter from services.homelab import kanal, pflege, protokoll, sammellauf, updates BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8")) T = datetime(2026, 9, 24, 20, 0, tzinfo=LOCAL_TZ).timestamp() ARCANE_KEY = "arc_Zz9TestSchluesselNurFuerDenTest" AUSFUEHRER_TOKEN = "kanal-token-nur-fuer-den-test-0123456789" GEHEIM = [ARCANE_KEY, AUSFUEHRER_TOKEN, "abcdefghijklmnop123", "geheim123", "AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw", "hunter2hunter", "supergeheim123456", "ghp_0123456789abcdefghijABCDEFGHIJ"] GEHEIME_AUSGABE = ( "\x1b[H\x1b[J\x1b[3J _______ __\nDownloading 10%\rDownloading 100%\n" "Authorization: Bearer abcdefghijklmnop123\ncurl https://tobi:geheim123@example.org/x\n" "TELEGRAM_BOT_TOKEN=123456789:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw\n" "POST https://api.telegram.org/bot123456789:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw/sendMessage\n" f"X-API-Key: {ARCANE_KEY}\nX-MC2-Ausfuehrer: {AUSFUEHRER_TOKEN}\npassword=hunter2hunter\n" "git clone https://ghp_0123456789abcdefghijABCDEFGHIJ@github.com/x/y\n" "E: Sub-process /usr/bin/dpkg returned an error code (1)") def _zeit(ts: float) -> str: return datetime.fromtimestamp(ts, LOCAL_TZ).strftime("%Y-%m-%d %H:%M:%S") def _auftrag(aid: str, aktion: str, vmid: int | None, status: str, code: int | None, text: str | None, ab: float, dauer: float = 3) -> dict: return {"id": aid, "aktion": aktion, "parameter": {} if vmid is None else {"vmid": vmid}, "status": status, "erstellt": ab, "abgeholt": None if status == "wartet" else ab + 1, "fertig": ab + 1 + dauer if status not in ("wartet", "laeuft") else None, "code": code, "text": text} def _lauf(lid: str, ziel: str, baustein: str, name: str, ende: float, **mehr) -> dict: return {"id": lid, "ziel": ziel, "baustein": baustein, "name": name, "status": "fertig", "start": ende - 60, "ende": ende, "ergebnis": None, "text": None, "schritte": ["snapshot: ok", "update: ok"], **mehr} @pytest.fixture def daten(tmp_path, monkeypatch): monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path)) monkeypatch.setenv("MC_NOTIFY_LOG", str(tmp_path / "notify.log")) for name in ("MC_AUSFUEHRER_TOKEN", "MC_ARCANE_KEY", "MC_ARCANE_ECHT", "MC_WAECHTER_STORE"): monkeypatch.delenv(name, raising=False) einstellungen_mod.einstellungen.cache_clear() monkeypatch.setattr(kanal, "_kontakt", {"zuletzt": None}) kanal.bericht_speichern(BERICHT) (tmp_path / "arcane.key").write_text(ARCANE_KEY + "\n", encoding="utf-8") (tmp_path / "ausfuehrer.token").write_text(AUSFUEHRER_TOKEN + "\n", encoding="utf-8") yield tmp_path einstellungen_mod.einstellungen.cache_clear() @pytest.fixture def geschichte(daten): """Ein Abend im Homelab, aus allen Quellen.""" kanal._json_schreiben(kanal._pfad(), [ _auftrag("a1", "suchen", 104, "fertig", 0, "Hit:1 http://deb.debian.org/debian trixie InRelease\n", T), _auftrag("a2", "snapshot", 104, "fertig", 0, "snapshot=mc2-20260924-192352\nfreeze guest filesystem", T + 10), _auftrag("a3", "update", 104, "fehler", 1, GEHEIME_AUSGABE, T + 20, dauer=60), _auftrag("a4", "bericht", None, "fertig", 0, json.dumps(BERICHT), T + 90), _auftrag("a5", "os_update", 105, "fehler", 2, "Abgelehnt: Der Ausführer läuft im Nur-Lesen-Modus.", T + 100), _auftrag("a6", "host_update", None, "verloren", None, "Der Ausführer hat nie geantwortet.", T + 110), _auftrag("a8", "sichern", 105, "fertig", 0, "INFO: fertig\nsicherung=local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst", T + 130), _auftrag("a9", "os_update", 101, "fertig", 0, "\n".join(f"Zeile {i}: " + "x" * 90 for i in range(200)), T + 140), _auftrag("a7", "suchen", 100, "wartet", None, None, T + 150), ]) for lauf in ( _lauf("l1", "ct-104", "app", "Gitea", T + 320, ergebnis="eingespielt", text="Gitea: eingespielt, Prüfung grün.", version_alt="1.27.2", version_neu="1.27.3", dringend=False, gemeldet=True), _lauf("l2", "ct-105", "os", "Proxmox Backup Server", T + 400, ergebnis="zurueckgerollt", dringend=True, text="Proxmox Backup Server: Update gescheitert (Die Weboberfläche antwortet nicht.) — automatisch die " "Sicherung von vorher zurückgespielt (vzdump-lxc-105.tar.zst). Läuft wieder wie vorher."), _lauf("l3", "ct-102", "app", "NetBird", T + 500, ergebnis="fehler", dringend=False, text="NetBird: Update nicht begonnen — snapshot gescheitert: kaputt. Am Gerät wurde nichts geändert."), _lauf("l4", "pve", "pakete", "Proxmox-Host", T + 600, ergebnis="eingespielt", pakete=5, sammellauf="s1", sammellauf_start=T + 550, dringend=False, gemeldet=False, text="Proxmox-Host: Pakete eingespielt. Ein Neustart ist ein eigener Knopf."), {**_lauf("l5", "ct-103", "os", "PVE Scripts Local", T + 650), "status": "unterbrochen"}, ): updates._merken(lauf) sammellauf._merken({"id": "s1", "status": "fertig", "start": T + 550, "ende": T + 700, "aktuell": None, "text": "1 Update eingespielt, alle Prüfungen grün.", "dringend": False, "meldung": "Homelab: 1 Update eingespielt, alle Prüfungen grün.", "schritte": [{"ziel": "pve", "name": "Proxmox-Host", "baustein": "pakete", "was": "5 Pakete", "status": "eingespielt", "lauf": "l4", "text": "Proxmox-Host: Pakete eingespielt."}]}) (daten / "mc2-waechter.json").write_text(json.dumps({ "hinweise": { "gast-platte:100": {"id": "gast-platte:100", "stufe": "gelb", "titel": "Platte von AdGuard Home zu 85 % voll", "text": "1,7 von 2,0 GB belegt (Container 100).", "seit": T + 900}, "ausfuehrer": {"id": "ausfuehrer", "stufe": "rot", "titel": "Der Ausführer auf dem Proxmox-Host schweigt", "text": "Seit über 30 Minuten kein Bericht.", "seit": T + 950}}, "verlauf": [ {"ts": T + 800, "art": "neu", "id": "gast:ct-104", "text": "Gitea antwortet nicht", "stufe": "rot"}, {"ts": T + 860, "art": "erledigt", "id": "gast:ct-104", "text": "Gitea antwortet nicht", "stufe": "rot"}, {"ts": T + 900, "art": "neu", "id": "gast-platte:100", "text": "Platte von AdGuard Home zu 85 % voll"}]}), encoding="utf-8") (daten / "notify.log").write_text("\n".join([ f"{_zeit(T + 320)} OK telegram direkt: Gitea: eingespielt, Prüfung grün.", (f"{_zeit(T + 400)} OK telegram direkt: Proxmox Backup Server: Update gescheitert (Die Weboberfläche " "antwortet nicht.) — automatisch die Sicherung von vorher zurückgespielt (vzdump-lxc-105.tar.zst). Läuft " "wieder wie vorher."), f"{_zeit(T + 700)} OK telegram direkt: Homelab: 1 Update eingespielt, alle Prüfungen grün.", (f"{_zeit(T + 800)} OK telegram direkt: Gitea antwortet nicht. Die Weboberfläche antwortet nicht. (Diese " "Meldung kam auch an Lucy.)"), f"{_zeit(T + 860)} QUEUED für Morgen-Digest: Erledigt: Gitea antwortet nicht. (Diese Meldung kam auch an Lucy.)", (f"{_zeit(T + 870)} FALLBACK (telegram fehlgeschlagen: ; Zweitweg: Telegram lehnt ab, HTTP-Fehler – Token " "oder Chat falsch?): Testmeldung aus den Einstellungen (Homelab, 24.09. 22:10 Uhr, Kennung abc123). Kommt sie " "an, funktioniert der Meldeweg."), f"{_zeit(T + 880)} OK telegram direkt: Guten Morgen, Commander. Heute Nacht gab es 2 Meldungen:", "- [Homelab-Update]: Gitea: eingespielt, Prüfung grün.", "- [Homelab-Problem]: Gitea antwortet nicht.", f"{_zeit(T + 890)} OK telegram direkt: Der Homelab-Teil läuft jetzt in Container 107. Nichts zu tun.", f"{_zeit(T + 895)} OK telegram direkt: Neuer Zugang: api_key=supergeheim123456", ]) + "\n", encoding="utf-8") pflege._schreiben({"nacht": "2026-09-24", "gaeste": { "104": {"name": "Gitea", "tag": "2026-09-24", "auftrag": None, "angestossen": T + 1000, "fehlschlaege": 0, "fehler": None}, "100": {"name": "AdGuard Home", "tag": "2026-09-24", "auftrag": None, "angestossen": T + 1010, "fehlschlaege": 2, "fehler": "E: Failed to fetch"}}, "ereignisse": [ {"ts": T + 1010, "vmid": 100, "name": "AdGuard Home", "art": "angestossen", "text": "Die Paketlisten sind vom 01.09.2026."}, {"ts": T + 1020, "vmid": 100, "name": "AdGuard Home", "art": "gescheitert", "text": "2. Fehlschlag in Folge: E: Failed to fetch"}]}) return daten def _nach_titel(eintraege: list[dict]) -> dict[str, dict]: return {e["titel"]: e for e in eintraege} def test_alle_quellen_mit_sprechenden_titeln(geschichte): eintraege = protokoll.eintraege() assert [e["zeit"] for e in eintraege] == sorted((e["zeit"] for e in eintraege), reverse=True) assert all(set(e) == {"zeit", "art", "titel", "text", "stufe", "ziel", "details"} for e in eintraege) assert {e["art"] for e in eintraege} == {"auftrag", "lauf", "sammellauf", "hinweis", "meldung", "pflege"} assert {e["stufe"] for e in eintraege} <= {"ok", "info", "warn", "fehler"} e = _nach_titel(eintraege) # Aufträge an den Ausführer: Aktion und Gerät im Titel, Status und Code im Text, Ausgabe als Einzelheiten. suchen = e["Paketlisten erneuert · Gitea"] assert (suchen["art"], suchen["stufe"], suchen["ziel"], suchen["text"]) == ("auftrag", "ok", "ct-104", "nach 3 s, Code 0") assert suchen["details"] == "Hit:1 http://deb.debian.org/debian trixie InRelease" assert e["Snapshot angelegt · Gitea"]["text"] == "mc2-20260924-192352, nach 3 s, Code 0" update = e["Update-Skript gescheitert · Gitea"] assert update["stufe"] == "fehler" assert update["text"] == "nach 1 min, Code 1: E: Sub-process /usr/bin/dpkg returned an error code (1)" assert e["Pakete einspielen abgelehnt · Proxmox Backup Server"]["stufe"] == "warn" host = e["Host-Pakete: keine Antwort · Proxmox-Host"] assert (host["stufe"], host["ziel"]) == ("fehler", "pve") assert e["Paketlisten werden erneuert · AdGuard Home"]["stufe"] == "info" assert e["Sicherung angelegt · Proxmox Backup Server"]["text"].startswith("vzdump-lxc-105-2026_09_24-22_10_00") lang = e["Pakete eingespielt · NPMplus"]["details"] assert lang.startswith("…\nZeile ") and len(lang) <= protokoll.DETAILS_MAX and lang.endswith("Zeile 199: " + "x" * 90) # Update-Läufe mit ihren Schritten assert e["Update eingespielt · Gitea 1.27.3"]["stufe"] == "ok" assert e["Update eingespielt · Gitea 1.27.3"]["details"] == "snapshot: ok\nupdate: ok" assert e["Paket-Update zurückgerollt · Proxmox Backup Server"]["stufe"] == "warn" assert e["Update nicht begonnen · NetBird"]["stufe"] == "warn" host_lauf = e["5 Host-Pakete eingespielt · Proxmox-Host"] assert "Teil von „Alle aktualisieren“." in host_lauf["details"] and "Sammelmeldung" in host_lauf["details"] assert e["Paket-Update unterbrochen · PVE Scripts Local"]["stufe"] == "fehler" # Sammellauf sl = e["Alle aktualisieren fertig · 1 von 1 eingespielt"] assert (sl["art"], sl["stufe"], sl["text"]) == ("sammellauf", "ok", "1 Update eingespielt, alle Prüfungen grün.") assert sl["details"] == "Proxmox-Host (5 Pakete): eingespielt – Proxmox-Host: Pakete eingespielt." # Hinweise des Wächters: Verlauf und, was dort fehlt, aus den aktuellen Hinweisen assert (e["Hinweis: Gitea antwortet nicht"]["stufe"], e["Hinweis: Gitea antwortet nicht"]["ziel"]) == ("fehler", "ct-104") assert e["Erledigt: Gitea antwortet nicht"]["stufe"] == "ok" platte = e["Hinweis: Platte von AdGuard Home zu 85 % voll"] assert (platte["stufe"], platte["ziel"], platte["text"]) == ("warn", "ct-100", "1,7 von 2,0 GB belegt (Container 100).") schweigt = e["Hinweis: Der Ausführer auf dem Proxmox-Host schweigt"] assert (schweigt["stufe"], schweigt["ziel"], schweigt["zeit"]) == ("fehler", "pve", T + 950) # Meldungen: Zeit, Betreff, ob raus assert [x["text"] for x in eintraege if x["titel"] == "Meldung raus · [Homelab-Update]"] == [ "Homelab: 1 Update eingespielt, alle Prüfungen grün.", "Gitea: eingespielt, Prüfung grün."] alarm = e["Meldung raus · [Alarm] Homelab-Update"] assert alarm["zeit"] == T + 400 and alarm["details"] == "Direkt an die Telegram-Bot-API gesendet." assert e["Meldung raus · [Homelab-Problem]"]["text"].startswith("Gitea antwortet nicht.") assert e["Meldung zurückgelegt · [Homelab wieder ok]"]["stufe"] == "info" test = e["Meldung ging nicht raus · [Test]"] assert test["stufe"] == "fehler" and test["details"].startswith("Telegram ging nicht: telegram fehlgeschlagen: ; " "Zweitweg: Telegram lehnt ab, HTTP-Fehler") morgen = e["Meldung raus · [Morgenmeldung Homelab]"] assert morgen["text"] == "Guten Morgen, Commander. Heute Nacht gab es 2 Meldungen:…" assert morgen["details"].endswith("- [Homelab-Problem]: Gitea antwortet nicht.") assert e["Meldung raus"]["text"] == "Der Homelab-Teil läuft jetzt in Container 107. Nichts zu tun." meldungen = [x for x in eintraege if x["art"] == "meldung"] assert len(meldungen) == 9 # die Folgezeilen der Morgenmeldung sind keine eigenen Meldungen # Pflege: Ereignisse, und aus älteren Ständen der letzte Anstoß assert e["Wöchentliche Suche gescheitert · AdGuard Home"]["stufe"] == "warn" gitea = e["Wöchentliche Suche angestoßen · Gitea"] assert (gitea["zeit"], gitea["text"], gitea["ziel"]) == (T + 1000, "Die Paketlisten waren älter als 7 Tage.", "ct-104") angestossen = [x for x in eintraege if x["titel"] == "Wöchentliche Suche angestoßen · AdGuard Home"] assert len(angestossen) == 1 and angestossen[0]["text"] == "Die Paketlisten sind vom 01.09.2026." def test_keine_geheimnisse_und_keine_steuerzeichen(geschichte): alles = json.dumps(protokoll.eintraege(berichte=True), ensure_ascii=False) for geheim in GEHEIM: assert geheim not in alles, geheim assert "\x1b" not in alles and "\\u001b" not in alles and "\r" not in alles and "\\r" not in alles details = _nach_titel(protokoll.eintraege())["Update-Skript gescheitert · Gitea"]["details"] assert "Downloading 100%" in details and "Downloading 10%" not in details assert "Authorization: ***" in details and "https://***@example.org/x" in details assert "TELEGRAM_BOT_TOKEN=***" in details and "/bot***/sendMessage" in details assert "X-API-Key: ***" in details and "X-MC2-Ausfuehrer: ***" in details and "password=***" in details assert "Neuer Zugang: api_key=***" in alles def test_schwaerzen_einzeln(daten): assert protokoll.schwaerzen(f"Schlüssel {ARCANE_KEY} benutzt") == "Schlüssel *** benutzt" assert protokoll.schwaerzen('{"api_key": "abc123def456"}') == '{"api_key": "***"}' assert protokoll.schwaerzen("Bearer eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N") == \ "Bearer ***" harmlos = "snapshot=mc2-20260924-192352 sha256:7304d54948626355e11a855dbcde49f58468c71e4cbf43b5315d3a2f5fd01a58" assert protokoll.schwaerzen(harmlos) == harmlos assert protokoll.schwaerzen(None) is None def test_berichte_nur_auf_wunsch_und_die_grenze(geschichte): ohne = protokoll.eintraege() assert not [x for x in ohne if x["titel"].startswith("Bericht")] mit = _nach_titel(protokoll.eintraege(berichte=True)) bericht = mit["Bericht geholt · Proxmox-Host"] assert bericht["details"] == "Proxmox VE 9.2.11, 7 Gäste, davon 7 laufend, 5 Paket-Updates auf dem Host." # Die Berichte, die der Ausführer von sich aus schickt (alle 10 min), sind keine Aufträge: kanal.py vermerkt sie. eingang = mit["Bericht empfangen · Proxmox-Host"] assert (eingang["art"], eingang["stufe"], eingang["ziel"]) == ("auftrag", "info", "pve") assert eingang["text"] == "Proxmox VE 9.2.11, 7 Gäste, davon 7 laufend, 5 Paket-Updates auf dem Host." kanal.bericht_speichern(BERICHT, aus_auftrag=True) # von einem Lauf geholt: steht schon als Auftrag da assert len(kanal.berichte()) == 1 drei = protokoll.eintraege(grenze=3) assert drei == ohne[:3] and drei[0]["titel"] == "Wöchentliche Suche gescheitert · AdGuard Home" assert len(protokoll.eintraege(grenze=0)) == 1 and len(protokoll.eintraege(grenze=10**6)) == len(ohne) def test_ohne_daten_leer(daten): (daten / "pve-bericht.json").unlink() assert protokoll.eintraege() == [] def test_schnittstelle(geschichte): app = FastAPI() from routers import homelab as homelab_router app.include_router(homelab_router.router) c = TestClient(app) antwort = c.get("/api/homelab/protokoll?grenze=2&berichte=1") assert antwort.status_code == 200 and len(antwort.json()["eintraege"]) == 2 alle = c.get("/api/homelab/protokoll").json()["eintraege"] assert len(alle) == len(protokoll.eintraege()) and not any(x["titel"].startswith("Bericht") for x in alle) assert any(x["titel"].startswith("Bericht") for x in c.get("/api/homelab/protokoll?berichte=1").json()["eintraege"]) # --- Kanal: längere Aufbewahrung -------------------------------------------------------------------------------- def test_kanal_behaelt_500_und_kuerzt_nur_alte_ausgaben(daten): alt = [_auftrag(f"x{i:03d}", "suchen", 104, "fertig", 0, f"{i:03d}" + "y" * 4997, T + i) for i in range(520)] kanal._json_schreiben(kanal._pfad(), alt) neu = kanal.anlegen("bericht") gespeichert = json.loads(kanal._pfad().read_text(encoding="utf-8")) fertig = [a for a in gespeichert if a["status"] == "fertig"] assert len(fertig) == kanal.BEHALTEN == 500 and fertig[0]["id"] == "x020" assert [a["id"] for a in gespeichert if a["status"] == "wartet"] == [neu] assert all(len(a["text"]) == kanal.TEXT_KURZ and a["text"].startswith("…") for a in fertig[:-kanal.VOLLTEXT]) assert all(len(a["text"]) == 5000 for a in fertig[-kanal.VOLLTEXT:]) def test_kanal_schreibt_beim_abholen_nur_bei_aenderung(daten, monkeypatch): schreiben = Counter() echt = kanal._json_schreiben monkeypatch.setattr(kanal, "_json_schreiben", lambda pfad, d: (schreiben.update([pfad.name]), echt(pfad, d))) assert kanal.naechster() is None and schreiben["ausfuehrer-auftraege.json"] == 0 aid = kanal.anlegen("suchen", {"vmid": 104}) assert kanal.naechster()["id"] == aid and schreiben["ausfuehrer-auftraege.json"] == 2 assert kanal.naechster() is None and schreiben["ausfuehrer-auftraege.json"] == 2 # --- Pflege: Ereignisse für das Protokoll ------------------------------------------------------------------------ NACHT = datetime(2026, 9, 25, 3, 0, tzinfo=LOCAL_TZ).timestamp() NAECHSTE_NACHT = datetime(2026, 9, 26, 3, 0, tzinfo=LOCAL_TZ).timestamp() def test_pflege_merkt_sich_anstoss_fehlschlag_und_erholung(daten, monkeypatch): stand = {"zuletzt": NACHT} monkeypatch.setattr(kanal, "zuletzt", lambda: stand["zuletzt"]) monkeypatch.setattr(waechter, "TROCKEN", False) def antworten(scheitert: set[int]) -> None: while (a := kanal.naechster()): if a["parameter"]["vmid"] in scheitert: kanal.ergebnis(a["id"], 100, "E: Failed to fetch http://security.debian.org") else: kanal.ergebnis(a["id"], 0, "Hit:1 http://deb.debian.org/debian bookworm InRelease") pflege.pruefe_paketlisten(jetzt=NACHT) antworten({100}) pflege.pruefe_paketlisten(jetzt=NACHT + 60) stand["zuletzt"] = NAECHSTE_NACHT pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT) antworten(set()) pflege.pruefe_paketlisten(jetzt=NAECHSTE_NACHT + 60) adguard = [(e["art"], e["text"]) for e in pflege.zustand()["ereignisse"] if e["vmid"] == 100] assert adguard == [("angestossen", "Die Paketlisten sind vom 22.10.2025."), ("gescheitert", "1. Fehlschlag in Folge: E: Failed to fetch http://security.debian.org"), ("angestossen", "Die Paketlisten sind vom 22.10.2025."), ("wieder_ok", "Die Paketlisten ließen sich wieder erneuern.")] # Anstöße tragen die (hier gespielte) Nacht, Ergebnisse die echte Zeit des Auftrags: nur die Menge vergleichen. titel = Counter(e["titel"] for e in protokoll.eintraege() if e["art"] == "pflege" and e["ziel"] == "ct-100") assert titel == Counter({"Wöchentliche Suche angestoßen · AdGuard Home": 2, "Wöchentliche Suche gescheitert · AdGuard Home": 1, "Wöchentliche Suche klappt wieder · AdGuard Home": 1})