HitonabiandClaude Opus 5.5
0df593c05a
Der erste Bericht nach dem Neustart vom 25.09. kam 34 s nach dem Hochfahren – Proxmox kannte die Gäste noch nicht („unknown“), und zehn Minuten lang zeigte die Oberfläche jeden Gast als unbekannt. Enthält ein Bericht solche Gäste, folgt der nächste nach 60 s. Der Grund steht jetzt auf Deutsch da. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
494 lines
25 KiB
Python
494 lines
25 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_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
|