280 lines
13 KiB
Python
280 lines
13 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"}
|
|
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)
|
|
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"
|
|
|
|
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 = {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
|