"""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"} monkeypatch.setattr(inventar, "neueste_version", lambda repo: { "tag": 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) 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" 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"] assert {b["id"]: b for b in z["ct-103"]["bausteine"]}["app"]["kurz"] == "0.5.8 → 1.2.1" 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 = z["vm-106"] assert arcane["name"] == "Arcane (Docker)" and arcane["bausteine"][0]["id"] == "freigabe" 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"] == "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_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]