Files
mission-control-v2/backend/tests/test_homelab.py
T

386 lines
18 KiB
Python

"""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 "MC_ARCANE_KEY" 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"] == "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]
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_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/<app>.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