HitonabiandClaude Opus 5.5
77ecb35a6d
Neuer Baustein „probe“ (POST /api/homelab/ziele/{id}/rueckweg-probe): legt den
Rückweg an wie vor einem Update, wertet die Prüfung absichtlich als rot, geht
zurück und prüft danach, ob das Gerät läuft und antwortet. Kein Update dabei;
Meldung grün bzw. dringend rot. Die Probe steht im Protokoll, nicht im
Update-Verlauf und nicht unter „Zuletzt eingespielt“. Der Snapshot bleibt
liegen (wie nach einem echten Rückweg).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
518 lines
26 KiB
Python
518 lines
26 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 "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/<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
|
|
|
|
|
|
|
|
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
|