HitonabiandClaude Opus 5.5
8ec261b8b2
- Sammellauf (services/homelab/sammellauf.py): alle Updates mit Knopf nacheinander über updates.starten, Gäste nach VMID, dann NPMplus und AdGuard, zuletzt die Host-Pakete (nie der Neustart). Rot stoppt den Rest (übersprungen, abgebrochen, dringende Meldung); grün gibt eine Sammelmeldung, die Erfolgsmeldungen der Schritte bleiben so lange aus. Nur einer gleichzeitig, der einzelne Knopf lehnt solange ab (START_SPERRE). Ein Neustart mitten im Lauf macht ihn beim Start zu „abgebrochen“, „unterbrochen (Neustart)“. - Protokoll (services/homelab/protokoll.py): Aufträge, Läufe, Sammelläufe, Wächter-Verlauf, Melde-Log und wöchentliches Suchen, neueste zuerst, ohne Geheimnisse. Der Kanal hebt 500 erledigte Aufträge auf und vermerkt eingehende Berichte; die Pflege führt eine kurze Ereignisliste, der Wächter-Verlauf die Stufe. - Einstellungen (services/homelab/einstellungen.py): Arcane-Schlüssel erst gegen Arcane prüfen, dann in arcane.key (0600) speichern, nie zurückgeben; echte Docker-Updates per Schalter. Die Umgebung geht vor. - Schnittstellen als eigener Block in routers/homelab.py vor den Ausführer-Endpunkten. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
336 lines
21 KiB
Python
336 lines
21 KiB
Python
"""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})
|