"""Homelab-Teil (Phase 3/4, 24.09.2026): Kanal zum Ausführer, Ziele aus dem echten Bericht des Proxmox-Hosts (tests/fixtures/pve-bericht.json, gelesen am 24.09.), „Jetzt updaten“ mit Rückweg.""" import importlib.util import json import os import threading import time from pathlib import Path import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from kern import einstellungen as einstellungen_mod from services.homelab import inventar, kanal, updates BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8")) AUSFUEHRER = Path(__file__).resolve().parents[2] / "deploy" / "homelab" / "ausfuehrer.py" @pytest.fixture def daten(tmp_path, monkeypatch): monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path)) monkeypatch.delenv("MC_AUSFUEHRER_TOKEN", raising=False) einstellungen_mod.einstellungen.cache_clear() monkeypatch.setattr(kanal, "_kontakt", {"zuletzt": None}) # Keine Netzabfragen: neueste Versionen und Weboberflächen sind fest. neueste = {"AdguardTeam/AdGuardHome": "0.107.79", "go-gitea/gitea": "1.27.3", "netbirdio/netbird": "0.79.0", "community-scripts/ProxmoxVE-Local": "1.2.1"} roh = {"community-scripts/ProxmoxVE-Local": "untagged-80028680f0b29a379726"} monkeypatch.setattr(inventar, "neueste_version", lambda repo: { "tag": neueste.get(repo, "2026-07-24-r1"), "tag_roh": roh.get(repo, neueste.get(repo, "2026-07-24-r1")), "version": neueste.get(repo, "2026-07-24-r1"), "datum": "2026-07-24"}) monkeypatch.setattr(inventar, "erreichbar", lambda url: True) monkeypatch.setattr(inventar.karenz, "skript_geaendert", lambda kennung: 0.0) # Update-Skripte alt, kein Netz monkeypatch.setattr(inventar.arcane, "app_version", lambda url: {"installiert": "2.12.0", "neu": "2.13.1", "update": True}) monkeypatch.delenv("MC_ARCANE_KEY", raising=False) monkeypatch.delenv("MC_ARCANE_ECHT", raising=False) yield tmp_path einstellungen_mod.einstellungen.cache_clear() # --- Kanal --------------------------------------------------------------------------------- def test_token_wird_einmal_erzeugt_und_geprueft(daten): wert = kanal.token() assert len(wert) > 30 and kanal.token() == wert assert kanal.token_gueltig(wert) and not kanal.token_gueltig("falsch") and not kanal.token_gueltig(None) if os.name == "posix": assert (daten / "ausfuehrer.token").stat().st_mode & 0o077 == 0 def test_auftraege_der_reihe_nach_und_verloren(daten, monkeypatch): erster = kanal.anlegen("bericht") zweiter = kanal.anlegen("suchen", {"vmid": 104}) with pytest.raises(ValueError): kanal.anlegen("rm -rf") assert kanal.naechster()["id"] == erster assert kanal.ergebnis(erster, 0, "ok") and not kanal.ergebnis(erster, 0, "doppelt") assert kanal.naechster() == {"id": zweiter, "aktion": "suchen", "parameter": {"vmid": 104}} monkeypatch.setattr(kanal, "VERLOREN_S", -1) assert kanal.naechster() is None assert kanal.auftrag(zweiter)["status"] == "verloren" def _client() -> TestClient: from routers import homelab app = FastAPI() app.include_router(homelab.router) return TestClient(app) def test_nur_der_ausfuehrer_darf_in_den_kanal(daten): c = _client() assert c.get("/api/homelab/ausfuehrer/auftrag").status_code == 403 assert c.post("/api/homelab/ausfuehrer/bericht", json=BERICHT, headers={"X-MC2-Ausfuehrer": "falsch"}).status_code == 403 kopf = {"X-MC2-Ausfuehrer": kanal.token()} assert c.post("/api/homelab/ausfuehrer/bericht", json=BERICHT, headers=kopf).json() == {"ok": True} assert c.get("/api/homelab/ausfuehrer/auftrag", headers=kopf).json() == {} assert c.get("/api/homelab/ausfuehrer").json()["verbunden"] is True # --- Ziele aus dem echten Bericht --------------------------------------------------------------- def _ziele(daten) -> dict[str, dict]: kanal.bericht_speichern(BERICHT) return {z["id"]: z for z in inventar.ziele()["ziele"]} def test_ziele_aus_dem_echten_bericht(daten): z = _ziele(daten) assert list(z) == ["pve", "ct-100", "ct-101", "ct-102", "ct-103", "ct-104", "ct-105", "vm-106"] host = {b["id"]: b for b in z["pve"]["bausteine"]} assert host["pakete"]["zustand"] == "neu" and host["pakete"]["aktion"]["pfad"] == "/api/homelab/ziele/pve/update" gitea = {b["id"]: b for b in z["ct-104"]["bausteine"]} assert (gitea["app"]["zustand"], gitea["app"]["kurz"]) == ("neu", "1.27.2 → 1.27.3") assert z["ct-104"]["name"] == "Gitea" and z["ct-104"]["rueckweg"] == "Snapshot vor jedem Update" assert (z["ct-104"]["rueckweg_art"], z["pve"]["rueckweg_art"]) == ("snapshot", "keiner") adguard = {b["id"]: b for b in z["ct-100"]["bausteine"]} assert adguard["app"]["zustand"] == "aktuell" # Paketlisten vom Oktober 2025: ehrlich „unbekannt“ mit Knopf zum Suchen statt „aktuell“. assert adguard["os"]["zustand"] == "unbekannt" and adguard["os"]["aktion"]["label"] == "Nach Updates suchen" pbs = {b["id"]: b for b in z["ct-105"]["bausteine"]} assert (pbs["os"]["zustand"], pbs["os"]["kurz"]) == ("neu", "58 Pakete") assert "Bind-Mount" in z["ct-105"]["rueckweg"] # PVE Scripts Local steht auf der neuesten (untagged) Veröffentlichung: aktuell, angezeigt als 1.2.1. pvesl = {b["id"]: b for b in z["ct-103"]["bausteine"]}["app"] assert (pvesl["zustand"], pvesl["installiert"], pvesl["aktion"]) == ("aktuell", "1.2.1", None) assert {b["id"]: b for b in z["ct-101"]["bausteine"]}["app"]["zustand"] == "aktuell" # Image jünger # NetBird hat keine Weboberfläche: erreichbar heißt hier „läuft“. assert z["ct-102"]["erreichbar"] is True arcane = {b["id"]: b for b in z["vm-106"]["bausteine"]} assert z["vm-106"]["name"] == "Arcane (Docker)" # Arcane läuft über seine eigene Schnittstelle: ohne Schlüssel sagt die Übersicht das ehrlich. assert (arcane["arcane"]["zustand"], arcane["arcane"]["kurz"]) == ("neu", "2.12.0 → 2.13.1") assert arcane["docker"]["zustand"] == "unbekannt" and "Einstellungen" in arcane["docker"]["grund"] def test_ohne_bericht_leer_und_ehrlich(daten): stand = inventar.ziele() assert stand["ziele"] == [] and stand["bericht"] is None and stand["ausfuehrer"]["verbunden"] is False def test_wachter_sieht_stumme_gaeste(daten, monkeypatch): from services import waechter kanal.bericht_speichern(BERICHT) monkeypatch.setattr(inventar, "erreichbar", lambda url: "153" not in url) befunde = waechter.pruefe_gaeste() assert [b.id for b in befunde] == ["gast:ct-104"] and befunde[0].titel == "Gitea antwortet nicht" assert waechter.pruefe_ausfuehrer() == [] # Bericht ist frisch # --- „Jetzt updaten“ mit einem nachgespielten Ausführer ---------------------------------------- def _ausfuehrer_spielen(stopp: threading.Event, neue_version: str, protokoll: list[str]) -> None: while not stopp.is_set(): auftrag = kanal.naechster() if not auftrag: time.sleep(0.02) continue protokoll.append(auftrag["aktion"]) text = "" if auftrag["aktion"] == "snapshot": text = "snapshot=mc2-20260924-190000" elif auftrag["aktion"] == "kernel_aufraeumen": text = "entfernt=20\nfrei=18300000000\nRemoving proxmox-kernel-6.17.2-1-pve-signed …" elif auftrag["aktion"] == "bericht": neu = json.loads(json.dumps(BERICHT)) next(g for g in neu["gaeste"] if g["vmid"] == 104)["app"]["version"] = neue_version text = json.dumps(neu) kanal.ergebnis(auftrag["id"], 0, text) @pytest.fixture def ausfuehrer(daten, monkeypatch): monkeypatch.setattr(updates, "WARTEN_NACH_UPDATE_S", 0) monkeypatch.setattr(kanal, "warten", _schnell_warten) meldungen: list[tuple[str, bool]] = [] monkeypatch.setattr(updates, "_melden", lambda text, dringend=False: meldungen.append((text, dringend))) verlauf: list[tuple] = [] monkeypatch.setattr(updates.update_verlauf, "eintragen", lambda *a: verlauf.append(a)) kanal.bericht_speichern(BERICHT) stopp, protokoll = threading.Event(), [] def starten(neue_version: str) -> None: threading.Thread(target=_ausfuehrer_spielen, args=(stopp, neue_version, protokoll), daemon=True).start() yield starten, protokoll, meldungen, verlauf stopp.set() _echtes_warten = kanal.warten def _schnell_warten(auftrag_id, zeitlimit_s, takt_s=2.0): return _echtes_warten(auftrag_id, min(zeitlimit_s, 10), 0.02) def _auf_ende(lauf_id: str) -> dict: ende = time.time() + 20 while time.time() < ende: lauf = next((x for x in updates.laeufe() if x["id"] == lauf_id), {}) if lauf.get("status") == "fertig": return lauf time.sleep(0.05) raise AssertionError(f"Lauf {lauf_id} wurde nicht fertig") def test_update_gruen(ausfuehrer, monkeypatch): starten, protokoll, meldungen, verlauf = ausfuehrer starten("1.27.3") antwort = updates.starten("ct-104", "app") assert antwort["ok"] assert updates.starten("ct-104", "app")["detail"] == "Für dieses Gerät läuft schon ein Update." lauf = _auf_ende(antwort["lauf"]) assert lauf["ergebnis"] == "eingespielt" and protokoll == ["snapshot", "update", "bericht"] assert meldungen == [("Gitea: eingespielt, Prüfung grün.", False)] assert verlauf[0][2:4] == ("ct-104", "eingespielt") def test_update_rot_geht_zurueck_und_meldet_dringend(ausfuehrer, monkeypatch): starten, protokoll, meldungen, _ = ausfuehrer monkeypatch.setattr(inventar, "erreichbar", lambda url: False) starten("1.27.3") lauf = _auf_ende(updates.starten("ct-104", "app")["lauf"]) assert lauf["ergebnis"] == "zurueckgerollt" assert protokoll == ["snapshot", "update", "bericht", "zurueck"] text, dringend = meldungen[0] assert dringend and "zurück auf den Snapshot mc2-20260924-190000" in text and "antwortet nicht" in text def test_update_ohne_neue_version_ist_rot(ausfuehrer): starten, _, _, _ = ausfuehrer starten("1.27.2") lauf = _auf_ende(updates.starten("ct-104", "app")["lauf"]) assert lauf["ergebnis"] == "zurueckgerollt" and "unverändert" in lauf["text"] def test_rueckweg_probe_gruen(ausfuehrer): starten, protokoll, meldungen, verlauf = ausfuehrer starten("1.27.2") lauf = _auf_ende(updates.starten("ct-104", "probe")["lauf"]) # Kein Update: Snapshot, zurück auf ihn, frischer Bericht — und die Probe steht nicht im Update-Verlauf. assert protokoll == ["snapshot", "zurueck", "bericht"] assert lauf["ergebnis"] == "probe-gruen" and "zurück auf den Snapshot mc2-20260924-190000" in lauf["text"] assert meldungen == [(lauf["text"], False)] and verlauf == [] def test_rueckweg_probe_rot_meldet_dringend(ausfuehrer, monkeypatch): starten, _, meldungen, _ = ausfuehrer monkeypatch.setattr(inventar, "erreichbar", lambda url: False) starten("1.27.2") lauf = _auf_ende(updates.starten("ct-104", "probe")["lauf"]) assert lauf["ergebnis"] == "fehler" and "Rückweg-Probe rot" in lauf["text"] and "antwortet nicht" in lauf["text"] assert meldungen[0][1] is True def test_rueckweg_probe_nur_mit_rueckweg(ausfuehrer): assert updates.starten("pve", "probe")["detail"].startswith("Für den Proxmox-Host") assert updates.starten("vm-106", "probe")["detail"].startswith("Dieses Gerät ist nicht freigegeben") def test_nicht_freigegeben_und_unbekannt(ausfuehrer): assert updates.starten("vm-106", "app")["detail"].startswith("Dieses Gerät ist nicht freigegeben") assert updates.starten("ct-999", "app")["ok"] is False assert updates.starten("ct-104", "boese")["ok"] is False # --- Der Ausführer selbst (läuft auf dem Proxmox-Host) ------------------------------------------- def _ausfuehrer_modul(): spec = importlib.util.spec_from_file_location("ausfuehrer", AUSFUEHRER) modul = importlib.util.module_from_spec(spec) spec.loader.exec_module(modul) return modul def test_ausfuehrer_entscheidet_selbst_ueber_etiketten(): a = _ausfuehrer_modul() assert a.erlaubt(["community-script", "git"]) and a.erlaubt(["watcher"]) assert not a.erlaubt(["community-script", "watcher-aus"]) and not a.erlaubt([]) def test_ausfuehrer_snapshot_faehigkeit(): a = _ausfuehrer_modul() arten = {"local-lvm": "lvmthin", "local": "dir"} assert a.snapshot_moeglich({"rootfs": "local-lvm:vm-104-disk-0,size=8G"}, arten) == (True, None) ok, grund = a.snapshot_moeglich({"rootfs": "local-lvm:vm-105-disk-0", "mp0": "/mnt/qnap,mp=/mnt/ds"}, arten) assert not ok and "Bind-Mount" in grund assert a.snapshot_moeglich({"rootfs": "local:105/vm-105-disk-0.raw"}, arten)[0] is False def test_ausfuehrer_lehnt_fremdes_ab(monkeypatch): a = _ausfuehrer_modul() monkeypatch.setattr(a, "_pvesh", lambda pfad, *p: [ {"vmid": 104, "type": "lxc", "tags": "community-script;git"}, {"vmid": 106, "type": "qemu", "tags": ""}]) ausgefuehrt: list[list[str]] = [] monkeypatch.setattr(a, "_laufen", lambda befehl, zeitlimit=60: ausgefuehrt.append(befehl) or (0, "ok")) assert a.ausfuehren({"aktion": "rm", "parameter": {}})["code"] == 2 assert "kein erlaubtes Etikett" in a.ausfuehren({"aktion": "update", "parameter": {"vmid": 106}})["text"] assert a.ausfuehren({"aktion": "zurueck", "parameter": {"vmid": 104, "snapshot": "vorher"}})["code"] == 2 assert a.ausfuehren({"aktion": "update", "parameter": {"vmid": "104; reboot"}})["code"] == 2 assert ausgefuehrt == [] assert a.ausfuehren({"aktion": "update", "parameter": {"vmid": 104}})["code"] == 0 assert ausgefuehrt[-1][:4] == ["pct", "exec", "104", "--"] and "PHS_SILENT=1" in ausgefuehrt[-1][-1] def test_docker_ueber_arcane_erst_als_probelauf(ausfuehrer, monkeypatch): _, _, meldungen, _ = ausfuehrer monkeypatch.setenv("MC_ARCANE_KEY", "schluessel") monkeypatch.setattr(inventar.arcane, "images_mit_update", lambda url: [ {"name": "ghcr.io/tobi/nerdquiz:latest", "aktuell": None, "neu": None, "umgebung": 0, "benutzt": ["nerdquiz"]}]) docker = {b["id"]: b for b in _ziele(None)["vm-106"]["bausteine"]}["docker"] assert (docker["zustand"], docker["kurz"], docker["aktion"]["label"]) == ("neu", "1 Images", "Probelauf") aufrufe: list[bool] = [] monkeypatch.setattr(updates.arcane, "updater", lambda url, probelauf: aufrufe.append(probelauf) or { "local": {"checked": 7, "updated": 0, "failed": 0, "items": [{"resourceName": "nerdquiz", "updateAvailable": True}]}}) lauf = _auf_ende(updates.starten("vm-106", "docker")["lauf"]) assert aufrufe == [True] and lauf["ergebnis"] == "offen" assert meldungen[-1][0] == "Arcane-Probelauf: 1 von 7 Containern würden aktualisiert (nerdquiz). Geändert wurde nichts." monkeypatch.delenv("MC_ARCANE_KEY") assert updates.starten("vm-106", "docker")["ok"] is False def test_docker_echt_mit_snapshot_nur_mit_etikett(ausfuehrer, monkeypatch): starten, protokoll, meldungen, _ = ausfuehrer monkeypatch.setenv("MC_ARCANE_KEY", "schluessel") monkeypatch.setenv("MC_ARCANE_ECHT", "1") monkeypatch.setattr(updates, "ARCANE_FRIST_S", 0) monkeypatch.setattr(inventar.arcane, "images_mit_update", lambda url: [ {"name": "nginx:alpine", "aktuell": None, "neu": None, "umgebung": 0, "benutzt": ["web"]}]) ergebnis = {"local": {"checked": 7, "updated": 1, "failed": 0, "items": []}} monkeypatch.setattr(updates.arcane, "updater", lambda url, probelauf: ergebnis) starten("1.27.3") # Ohne Etikett (wie VM 106 heute): kein Snapshot — Update-Liste und Rückfrage sagen es, statt einen zu versprechen. arcane_ziel = _ziele(None)["vm-106"] docker = {b["id"]: b for b in arcane_ziel["bausteine"]}["docker"] assert arcane_ziel["rueckweg_art"] == "keiner" and "Etikett „watcher“ fehlt" in docker["aktion"]["frage"] lauf = _auf_ende(updates.starten("vm-106", "docker")["lauf"]) assert (lauf["ergebnis"], protokoll) == ("eingespielt", []) # Mit Etikett: vorher ein Snapshot der VM; ist die Prüfung rot, geht es darauf zurück. bericht = json.loads(json.dumps(BERICHT)) next(g for g in bericht["gaeste"] if g["vmid"] == 106)["erlaubt"] = True kanal.bericht_speichern(bericht) arcane_ziel = next(z for z in inventar.ziele()["ziele"] if z["id"] == "vm-106") docker = {b["id"]: b for b in arcane_ziel["bausteine"]}["docker"] assert arcane_ziel["rueckweg_art"] == "snapshot" and "Vorher wird ein Snapshot angelegt" in docker["aktion"]["frage"] ergebnis["local"]["failed"] = 1 lauf = _auf_ende(updates.starten("vm-106", "docker")["lauf"]) assert (lauf["ergebnis"], protokoll) == ("zurueckgerollt", ["snapshot", "zurueck"]) text, dringend = meldungen[-1] assert dringend and "zurück auf den Snapshot mc2-20260924-190000" in text and "1 Container" in text # Grün: Der Snapshot hat seinen Zweck erfüllt und geht gleich wieder weg (die VM schreibt laufend). ergebnis["local"]["failed"] = 0 lauf = _auf_ende(updates.starten("vm-106", "docker")["lauf"]) assert lauf["ergebnis"] == "eingespielt" and protokoll[2:] == ["snapshot", "snapshot_loeschen"] assert meldungen[-1] == ("Arcane: 1 Container aktualisiert, Prüfung grün.", False) def test_ausfuehrer_nur_lesen(monkeypatch): a = _ausfuehrer_modul() monkeypatch.setattr(a, "NUR_LESEN", True) monkeypatch.setattr(a, "_pvesh", lambda pfad, *p: [{"vmid": 104, "type": "lxc", "tags": "community-script"}]) ausgefuehrt: list[list[str]] = [] monkeypatch.setattr(a, "_laufen", lambda befehl, zeitlimit=60: ausgefuehrt.append(befehl) or (0, "ok")) antwort = a.ausfuehren({"aktion": "update", "parameter": {"vmid": 104}}) assert antwort["code"] == 2 and "Nur-Lesen" in antwort["text"] and ausgefuehrt == [] monkeypatch.setattr(a, "bericht", lambda: {"gaeste": []}) assert a.ausfuehren({"aktion": "bericht", "parameter": {}})["code"] == 0 def test_eigener_container_erscheint_nicht(daten): bericht = json.loads(json.dumps(BERICHT)) bericht["gaeste"].append({"vmid": 107, "art": "lxc", "name": "homelab-orchestrator", "status": "running", "etiketten": ["mc2"], "erlaubt": False, "ip": "192.168.178.31"}) kanal.bericht_speichern(bericht) assert "ct-107" not in {z["id"] for z in inventar.ziele()["ziele"]} def test_strom_stoesst_bei_aenderung_an_und_meldet_sich(daten): import asyncio from routers import homelab as homelab_router class Anfrage: def __init__(self): self.runden = 0 async def is_disconnected(self): self.runden += 1 return self.runden > 4 async def lesen() -> list[str]: teile = [] async for teil in homelab_router._strom(Anfrage(), takt_s=0.01, lebenszeichen_takte=2): teile.append(teil) if len(teile) == 1: kanal.bericht_speichern(BERICHT) # ein neuer Bericht → Anstoß „homelab“ return teile teile = asyncio.run(lesen()) assert teile[0] == ": verbunden\n\n" assert teile[1] == 'event: invalidate\ndata: {"keys": ["homelab"]}\n\n' assert 'event: invalidate\ndata: {"keys": []}\n\n' in teile[2:] # Lebenszeichen def test_hinweise_des_homelab_waechters(daten, monkeypatch): from services import waechter monkeypatch.setattr(waechter, "STORE_PATH", daten / "mc2-waechter.json") (daten / "mc2-waechter.json").write_text(json.dumps({"hinweise": {"gast:ct-104": { "id": "gast:ct-104", "stufe": "rot", "titel": "Gitea antwortet nicht", "text": "…", "seit": 1}}}), encoding="utf-8") antwort = _client().get("/api/homelab/hinweise").json() assert [h["titel"] for h in antwort["hinweise"]] == ["Gitea antwortet nicht"] def test_apps_mit_eigenem_updater_bekommen_keinen_knopf(daten, monkeypatch): """AdGuard und PVE Scripts Local aktualisiert das Community-Script nicht (ct/.sh, 24.09.2026).""" bericht = json.loads(json.dumps(BERICHT)) next(g for g in bericht["gaeste"] if g["vmid"] == 100)["app"]["version"] = "0.107.70" kanal.bericht_speichern(bericht) adguard = {b["id"]: b for b in {z["id"]: z for z in inventar.ziele()["ziele"]}["ct-100"]["bausteine"]}["app"] assert adguard["zustand"] == "neu" and adguard["aktion"] is None and "eigene Oberfläche" in adguard["grund"] antwort = updates.starten("ct-103", "app") assert antwort["ok"] is False and "Community-Script" in antwort["detail"] def test_ausfuehrer_erkennt_beide_formate_von_update(): """Nach dem ersten echten Update (Gitea, 24.09.2026) schrieb das Community-Script /usr/bin/update neu.""" a = _ausfuehrer_modul() alt = 'bash -c "$(curl -fsSL https://raw.githubusercontent.com/community-scripts/ProxmoxVE/main/ct/adguard.sh)"' neu = ('#!/usr/bin/env bash\nexport SCRIPT_SLUG="gitea"\nexport UPDATE_SCRIPT_NAME="gitea"\n' 'bash -c "$(curl -fsSL "${COMMUNITY_SCRIPTS_URL}/ct/${UPDATE_SCRIPT_NAME}.sh")"\n') assert a.app_kennung_aus(alt) == "adguard" assert a.app_kennung_aus(neu) == "gitea" assert a.app_kennung_aus("") is None def test_waechter_und_uebersicht_sehen_volle_gastplatten(daten): """24.09.2026: AdGuard stand bei 100 % seiner 2 GB — jetzt sagen es Wächter und Geräteliste.""" from services import waechter bericht = json.loads(json.dumps(BERICHT)) for g in bericht["gaeste"]: if g["vmid"] == 100: g["platte"] = {"belegt": 2_000_000_000, "gesamt": 2_040_000_000} if g["vmid"] == 104: g["platte"] = {"belegt": 1_700_000_000, "gesamt": 2_000_000_000} if g["vmid"] == 101: g["platte"] = {"belegt": 500_000_000, "gesamt": 2_000_000_000} kanal.bericht_speichern(bericht) befunde = {b.id: b for b in waechter.pruefe_gast_platten()} assert set(befunde) == {"gast-platte:100", "gast-platte:104"} assert (befunde["gast-platte:100"].stufe, befunde["gast-platte:104"].stufe) == ("rot", "gelb") assert befunde["gast-platte:100"].titel == "Platte von AdGuard Home zu 98 % voll" assert "2,0 von 2,0 GB" in befunde["gast-platte:100"].text z = {z["id"]: z for z in inventar.ziele()["ziele"]} assert z["ct-100"]["platte"] == {"belegt": 2_000_000_000, "gesamt": 2_040_000_000} and z["pve"]["platte"] is None def test_images_ohne_container_sind_kein_update(daten, monkeypatch): """25.09.2026: Nach dem ersten echten Lauf blieben 9 Images „neu“ — Basis-Images, die kein Container benutzt.""" monkeypatch.setenv("MC_ARCANE_KEY", "schluessel") ohne = [{"name": n, "aktuell": None, "neu": None, "umgebung": 0, "benutzt": []} for n in ("alpine:latest", "node:22")] monkeypatch.setattr(inventar.arcane, "images_mit_update", lambda url: ohne) docker = {b["id"]: b for b in _ziele(None)["vm-106"]["bausteine"]}["docker"] assert (docker["zustand"], docker["aktion"]) == ("aktuell", None) assert docker["grund"].startswith("2 ältere Images ohne Container (alpine:latest, node:22)") genutzt = {"name": "ghcr.io/tobi/nerdquiz:latest", "aktuell": None, "neu": None, "umgebung": 0, "benutzt": ["quiz"]} monkeypatch.setattr(inventar.arcane, "images_mit_update", lambda url: [*ohne, genutzt]) docker = {b["id"]: b for b in _ziele(None)["vm-106"]["bausteine"]}["docker"] assert (docker["zustand"], docker["kurz"]) == ("neu", "1 Images") def test_neuer_kernel_heisst_neustart(daten): """25.09.2026: Proxmox legt bei Kernel-Updates kein reboot-required an — der Ausführer vergleicht selbst.""" a = _ausfuehrer_modul() vorhanden = ["6.8.12-4-pve", "7.0.14-12-pve", "7.0.14-19-pve"] assert a.neuer_kernel("7.0.14-12-pve", vorhanden) == "7.0.14-19-pve" assert a.neuer_kernel("7.0.14-19-pve", vorhanden) is None assert a.neuer_kernel("7.0.14-12-pve", vorhanden, pin="7.0.14-12-pve") is None # festgepinnt assert a.neuer_kernel(None, vorhanden) is None and a.neuer_kernel("7.0.14-12-pve", []) is None bericht = json.loads(json.dumps(BERICHT)) bericht["host"].update(neustart_noetig=True, kernel="7.0.14-12-pve", kernel_neu="7.0.14-19-pve") kanal.bericht_speichern(bericht) host = {b["id"]: b for b in next(z for z in inventar.ziele()["ziele"] if z["id"] == "pve")["bausteine"]} assert host["neustart"]["grund"] == "Neuer Kernel 7.0.14-19-pve ist installiert, es läuft noch 7.0.14-12-pve." assert host["neustart"]["aktion"]["label"] == "Jetzt neu starten" def test_alte_kernel_nur_auf_knopfdruck(ausfuehrer): """25.09.2026: 25 Kernel auf dem Host, 20 davon hielt apt für entbehrlich.""" starten, protokoll, meldungen, verlauf = ausfuehrer starten("1.27.3") lauf = _auf_ende(updates.starten("pve", "kernel")["lauf"]) assert (lauf["ergebnis"], protokoll) == ("aufgeraeumt", ["kernel_aufraeumen", "bericht"]) assert meldungen[-1] == ("Proxmox-Host: 20 alte Kernel entfernt, 18,3 GB frei geworden.", False) assert verlauf[-1][2:4] == ("pve", "aufgeraeumt") from services.homelab import protokoll as prot assert prot._lauf(lauf, lambda *a: None) is None or True # Titel-Zweig existiert (kein KeyError) def test_ausfuehrer_waehlt_alte_kernel_selbst(): a = _ausfuehrer_modul() simulation = """NOTE: This is only a simulation! Reading package lists... The following packages will be REMOVED: proxmox-kernel-6.14.11-2-pve-signed proxmox-kernel-7.0.14-12-pve-signed libfoo1 Remv proxmox-kernel-6.14.11-2-pve-signed [6.14.11-2] Remv proxmox-kernel-7.0.14-12-pve-signed [7.0.14-12] Remv proxmox-kernel-7.0.14-5-pve-signed [7.0.14-5] Remv proxmox-kernel-6.17 [6.17.13-21] Remv libfoo1 [1.2-3] Remv pve-kernel-6.8.12-4-pve [6.8.12-4] """ behalten = {"7.0.14-12-pve", "7.0.14-19-pve"} assert a.alte_kernel(simulation, behalten) == [ "proxmox-kernel-6.14.11-2-pve-signed", "proxmox-kernel-7.0.14-5-pve-signed", "pve-kernel-6.8.12-4-pve"] def test_nach_dem_hochfahren_bald_wieder_berichten(): """25.09.2026: Der erste Bericht nach dem Neustart kam 34 s nach dem Hochfahren — alle Gäste „unknown“.""" a = _ausfuehrer_modul() assert a.naechster_bericht_in({"gaeste": [{"status": "running"}, {"status": "stopped"}]}) == a.BERICHT_ALLE_S assert a.naechster_bericht_in({"gaeste": [{"status": "running"}, {"status": "unknown"}]}) == a.BERICHT_BALD_S == 60 assert a.naechster_bericht_in({}) == a.BERICHT_ALLE_S