HitonabiandClaude Opus 5.5
54a3649d5e
User 25.09.: Die Karten waren „immer da“ und ließen sich weder bestätigen noch wegklicken. Beendete Aufträge der KI-Box und das letzte „Alle aktualisieren“ im Homelab haben jetzt einen Knopf „Ausblenden“ (quittiert wird auf dem Server, gilt also auf allen Geräten; das Protokoll behält alles). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
400 lines
22 KiB
Python
400 lines
22 KiB
Python
"""„Alle aktualisieren“ (services/homelab/sammellauf.py, 24.09.2026): Reihenfolge, Stopp bei Rot, Sperre, Neustart
|
||
mitten im Lauf und zurückgehaltene Einzelmeldungen. Mit einem nachgespielten Ausführer und dem echten Bericht vom
|
||
24.09. (tests/fixtures/pve-bericht.json); NPMplus und AdGuard haben hier zusätzlich Paket-Updates."""
|
||
|
||
import json
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
from kern import einstellungen as einstellungen_mod
|
||
from services.homelab import inventar, kanal, karenz, sammellauf, updates
|
||
|
||
BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8"))
|
||
SPERRE = "Es läuft gerade ‚Alle aktualisieren‘."
|
||
NEU = {102: "0.79.0", 104: "1.27.3"} # NetBird und Gitea nach ihrem Update
|
||
ALLE = [("ct-102", "app"), ("ct-104", "app"), ("ct-105", "os"), ("ct-101", "os"), ("ct-100", "os"), ("pve", "pakete")]
|
||
|
||
|
||
def _bericht(versionen: dict[int, str] | None = None) -> dict:
|
||
"""Der echte Bericht mit frischen Paketlisten: PBS 58, NPMplus 3 und AdGuard 2 Pakete."""
|
||
neu = json.loads(json.dumps(BERICHT))
|
||
jetzt = int(time.time())
|
||
for g in neu["gaeste"]:
|
||
if g["vmid"] == 101:
|
||
g["os_updates"] = {"anzahl": 3, "listen_stand": jetzt}
|
||
elif g["vmid"] == 100:
|
||
g["os_updates"] = {"anzahl": 2, "listen_stand": jetzt}
|
||
elif g["vmid"] == 105:
|
||
g["os_updates"]["listen_stand"] = jetzt
|
||
if versionen and g["vmid"] in versionen:
|
||
g["app"]["version"] = versionen[g["vmid"]]
|
||
return neu
|
||
|
||
|
||
class Ausfuehrer:
|
||
"""Nachgespielt: beantwortet Aufträge der Reihe nach. Nach „update“ meldet der nächste Bericht die neue
|
||
App-Version — außer für VMIDs in `unveraendert` (dann ist die Prüfung rot). `weiter` hält ihn an;
|
||
`fehlschlag` (Aktion, VMID) → (Code, Text)."""
|
||
|
||
def __init__(self) -> None:
|
||
self.protokoll: list[tuple[str, int | None]] = []
|
||
self.weiter, self.stopp = threading.Event(), threading.Event()
|
||
self.weiter.set()
|
||
self.unveraendert: set[int] = set()
|
||
self.aktualisiert: set[int] = set()
|
||
self.fehlschlag: dict[tuple[str, int | None], tuple[int, str]] = {}
|
||
threading.Thread(target=self._laufen, daemon=True).start()
|
||
|
||
def _laufen(self) -> None:
|
||
while not self.stopp.is_set():
|
||
if not self.weiter.is_set():
|
||
time.sleep(0.01)
|
||
continue
|
||
a = kanal.naechster()
|
||
if not a:
|
||
time.sleep(0.01)
|
||
continue
|
||
vmid = a["parameter"].get("vmid")
|
||
self.protokoll.append((a["aktion"], vmid))
|
||
code, text = self.fehlschlag.get((a["aktion"], vmid), (0, ""))
|
||
if code == 0 and a["aktion"] == "snapshot":
|
||
text = "snapshot=mc2-20260924-220000"
|
||
elif code == 0 and a["aktion"] == "update":
|
||
self.aktualisiert.add(vmid)
|
||
elif code == 0 and a["aktion"] == "bericht":
|
||
text = json.dumps(_bericht({v: NEU[v] for v in self.aktualisiert - self.unveraendert if v in NEU}))
|
||
kanal.ergebnis(a["id"], code, text)
|
||
|
||
|
||
@pytest.fixture
|
||
def homelab(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path))
|
||
for name in ("MC_AUSFUEHRER_TOKEN", "MC_ARCANE_KEY", "MC_ARCANE_ECHT"):
|
||
monkeypatch.delenv(name, raising=False)
|
||
einstellungen_mod.einstellungen.cache_clear()
|
||
monkeypatch.setattr(kanal, "_kontakt", {"zuletzt": None})
|
||
neueste = {"netbirdio/netbird": "0.79.0", "go-gitea/gitea": "1.27.3", "AdguardTeam/AdGuardHome": "0.107.79",
|
||
"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(karenz, "skript_geaendert", lambda kennung: 0.0)
|
||
monkeypatch.setattr(inventar.arcane, "app_version",
|
||
lambda url: {"installiert": "2.12.0", "neu": "2.13.1", "update": True})
|
||
monkeypatch.setattr(updates, "WARTEN_NACH_UPDATE_S", 0)
|
||
echt_warten = kanal.warten
|
||
monkeypatch.setattr(kanal, "warten", lambda aid, zeitlimit_s, takt_s=2.0: echt_warten(aid, min(zeitlimit_s, 10), 0.02))
|
||
monkeypatch.setattr(sammellauf, "TAKT_S", 0.02)
|
||
meldungen: list[tuple[str, bool]] = []
|
||
|
||
def melden(text: str, dringend: bool = False) -> None:
|
||
meldungen.append((text, dringend))
|
||
|
||
monkeypatch.setattr(updates, "_melden", melden)
|
||
monkeypatch.setattr(sammellauf, "_melden", melden)
|
||
verlauf: list[tuple] = []
|
||
monkeypatch.setattr(updates.update_verlauf, "eintragen", lambda *a: verlauf.append(a))
|
||
kanal.bericht_speichern(_bericht())
|
||
ausfuehrer = Ausfuehrer()
|
||
yield SimpleNamespace(ausfuehrer=ausfuehrer, meldungen=meldungen, verlauf=verlauf, daten=tmp_path)
|
||
ausfuehrer.weiter.set()
|
||
ende = time.time() + 20 # keinen Faden in den nächsten Test mitnehmen
|
||
while (sammellauf.laeuft() or updates.laufende_ziele()) and time.time() < ende:
|
||
time.sleep(0.05)
|
||
ausfuehrer.stopp.set()
|
||
einstellungen_mod.einstellungen.cache_clear()
|
||
|
||
|
||
def _auf_ende(zeitlimit: float = 30) -> dict:
|
||
ende = time.time() + zeitlimit
|
||
while time.time() < ende:
|
||
sl = sammellauf.aktueller()
|
||
if sl and sl["status"] != "laeuft":
|
||
return sl
|
||
time.sleep(0.05)
|
||
raise AssertionError("„Alle aktualisieren“ wurde nicht fertig")
|
||
|
||
|
||
def _auf_lauf(lauf_id: str) -> dict:
|
||
ende = time.time() + 20
|
||
while time.time() < ende:
|
||
lauf = updates.lauf_lesen(lauf_id)
|
||
if lauf and lauf["status"] != "laeuft":
|
||
return lauf
|
||
time.sleep(0.05)
|
||
raise AssertionError(f"Lauf {lauf_id} wurde nicht fertig")
|
||
|
||
|
||
# --- Plan -----------------------------------------------------------------------------------------------
|
||
|
||
def test_plan_reihenfolge_nadeloehre_zuletzt_und_der_host_ganz_am_ende(homelab):
|
||
plan = sammellauf.plan()
|
||
assert [(s["ziel"], s["baustein"]) for s in plan["schritte"]] == ALLE
|
||
assert plan["schritte"][0] == {"ziel": "ct-102", "name": "NetBird", "baustein": "app", "was": "0.77.1 → 0.79.0",
|
||
"rueckweg_art": "snapshot"}
|
||
assert [(s["name"], s["was"], s["rueckweg_art"]) for s in plan["schritte"][2:]] == [
|
||
("Proxmox Backup Server", "58 Pakete", "keiner"), ("NPMplus", "3 Pakete", "snapshot"),
|
||
("AdGuard Home", "2 Pakete", "snapshot"), ("Proxmox-Host", "5 Pakete", "keiner")]
|
||
assert plan["hinweis"] == ("Ohne Rückweg: Proxmox Backup Server – scheitert dort das Update, geht es nicht von "
|
||
"selbst zurück. Der Proxmox-Host kommt zuletzt und hat keinen Rückweg; neu gestartet "
|
||
"wird er nicht.")
|
||
|
||
|
||
def test_plan_nur_mit_knopf_und_nie_der_neustart(homelab, monkeypatch):
|
||
bericht = _bericht()
|
||
bericht["host"]["neustart_noetig"] = True
|
||
kanal.bericht_speichern(bericht)
|
||
# Gitea wartet nach einer frischen Skriptänderung: „neu“, aber ohne Knopf — also nicht dabei.
|
||
monkeypatch.setattr(karenz, "skript_geaendert", lambda kennung: time.time() - 3600 if kennung == "gitea" else 0.0)
|
||
ziele = [(s["ziel"], s["baustein"]) for s in sammellauf.plan()["schritte"]]
|
||
assert ziele == [z for z in ALLE if z != ("ct-104", "app")] and ("pve", "neustart") not in ziele
|
||
|
||
|
||
def test_plan_docker_nur_wenn_echt(homelab, monkeypatch):
|
||
monkeypatch.setenv("MC_ARCANE_KEY", "schluessel-fuer-den-test")
|
||
monkeypatch.setattr(inventar.arcane, "images_mit_update", lambda url: [
|
||
{"name": "ghcr.io/tobi/nerdquiz:latest", "aktuell": None, "neu": None, "umgebung": 0, "benutzt": ["nerdquiz"]}])
|
||
plan = sammellauf.plan()
|
||
assert "vm-106" not in [s["ziel"] for s in plan["schritte"]]
|
||
assert plan["hinweis"].endswith("Die Docker-Images (Arcane) sind nicht dabei, solange Docker-Updates nur als "
|
||
"Probelauf laufen.")
|
||
monkeypatch.setenv("MC_ARCANE_ECHT", "1")
|
||
plan = sammellauf.plan()
|
||
assert [s["ziel"] for s in plan["schritte"]] == ["ct-102", "ct-104", "ct-105", "vm-106", "ct-101", "ct-100", "pve"]
|
||
docker = plan["schritte"][3]
|
||
assert (docker["baustein"], docker["was"], docker["rueckweg_art"]) == ("docker", "1 Images", "keiner")
|
||
assert "Ohne Rückweg: Proxmox Backup Server und Arcane (Docker)" in plan["hinweis"]
|
||
|
||
|
||
# --- Ablauf ---------------------------------------------------------------------------------------------
|
||
|
||
def test_gruen_nacheinander_mit_einer_sammelmeldung(homelab):
|
||
antwort = sammellauf.starten()
|
||
assert antwort["ok"] and antwort["sammellauf"]["status"] == "laeuft"
|
||
assert set(antwort["sammellauf"]) == {"id", "status", "start", "ende", "aktuell", "schritte", "text"}
|
||
assert set(antwort["sammellauf"]["schritte"][0]) == {"ziel", "name", "baustein", "was", "status", "lauf", "text"}
|
||
sl = _auf_ende()
|
||
assert sl["status"] == "fertig" and sl["aktuell"] is None and sl["ende"] >= sl["start"]
|
||
assert [s["status"] for s in sl["schritte"]] == ["eingespielt"] * 6
|
||
assert sl["text"] == "6 Updates eingespielt, alle Prüfungen grün."
|
||
assert homelab.ausfuehrer.protokoll == [
|
||
("snapshot", 102), ("update", 102), ("bericht", None),
|
||
("snapshot", 104), ("update", 104), ("bericht", None),
|
||
("os_update", 105), ("bericht", None), # PBS: weder Snapshot noch Sicherung
|
||
("snapshot", 101), ("os_update", 101), ("bericht", None),
|
||
("snapshot", 100), ("os_update", 100), ("bericht", None),
|
||
("host_update", None), ("bericht", None)]
|
||
# Die Erfolgsmeldungen der Schritte bleiben aus; es kommt genau die eine Sammelmeldung.
|
||
assert homelab.meldungen == [("Homelab: 6 Updates eingespielt, alle Prüfungen grün.", False)]
|
||
laeufe = {x["id"]: x for x in updates.laeufe(20)}
|
||
assert all(laeufe[s["lauf"]]["gemeldet"] is False and laeufe[s["lauf"]]["sammellauf"] == sl["id"]
|
||
for s in sl["schritte"])
|
||
assert laeufe[sl["schritte"][1]["lauf"]]["version_neu"] == "1.27.3"
|
||
# Im Update-Verlauf sind die Schritte EIN Lauf mit eigenem Anlass.
|
||
assert len(homelab.verlauf) == 6 and len({v[0] for v in homelab.verlauf}) == 1
|
||
assert {v[1] for v in homelab.verlauf} == {"Alle aktualisieren"}
|
||
assert [v[2:4] for v in homelab.verlauf][:2] == [("ct-102", "eingespielt"), ("ct-104", "eingespielt")]
|
||
|
||
|
||
def test_rot_stoppt_und_meldet_dringend(homelab):
|
||
homelab.ausfuehrer.unveraendert.add(104) # Gitea: die Version bleibt → Prüfung rot → zurück
|
||
assert sammellauf.starten()["ok"]
|
||
sl = _auf_ende()
|
||
assert sl["status"] == "abgebrochen" and sl["text"] == "Abgebrochen bei Gitea: Update zurückgerollt."
|
||
assert [s["status"] for s in sl["schritte"]] == ["eingespielt", "zurueckgerollt"] + ["uebersprungen"] * 4
|
||
assert sl["schritte"][2]["text"] == "Nicht mehr gelaufen: „Alle aktualisieren“ wurde abgebrochen."
|
||
assert homelab.ausfuehrer.protokoll[-1] == ("zurueck", 104)
|
||
assert not {v for _, v in homelab.ausfuehrer.protokoll} & {105, 101, 100}
|
||
assert ("host_update", None) not in homelab.ausfuehrer.protokoll
|
||
(gitea, dringend_1), (sammel, dringend_2) = homelab.meldungen # NetBirds Erfolg bleibt still
|
||
assert dringend_1 and gitea.startswith("Gitea: Update gescheitert") and "zurück auf den Snapshot" in gitea
|
||
assert dringend_2 and sammel == ("Homelab: „Alle aktualisieren“ abgebrochen bei Gitea (Update zurückgerollt). "
|
||
"Vorher eingespielt: NetBird. Nicht mehr gelaufen: Proxmox Backup Server, "
|
||
"NPMplus, AdGuard Home und Proxmox-Host.")
|
||
assert not sammellauf.laeuft() and not updates.laufende_ziele()
|
||
|
||
|
||
def test_nicht_begonnen_stoppt_auch_und_die_fehlermeldung_bleibt(homelab):
|
||
homelab.ausfuehrer.fehlschlag[("snapshot", 102)] = (1, "snapshot feature is not available")
|
||
assert sammellauf.starten()["ok"]
|
||
sl = _auf_ende()
|
||
assert [s["status"] for s in sl["schritte"]] == ["fehler"] + ["uebersprungen"] * 5
|
||
assert homelab.ausfuehrer.protokoll == [("snapshot", 102)]
|
||
assert homelab.meldungen[0] == (("NetBird: Update nicht begonnen — snapshot gescheitert: snapshot feature is "
|
||
"not available. Am Gerät wurde nichts geändert."), False)
|
||
assert homelab.meldungen[1][1] is True and "abgebrochen bei NetBird (Update gescheitert)" in homelab.meldungen[1][0]
|
||
|
||
|
||
def test_abgelehnter_schritt_wird_uebersprungen_und_es_geht_weiter(homelab, monkeypatch):
|
||
gesperrt = {"an": False}
|
||
monkeypatch.setattr(karenz, "skript_geaendert",
|
||
lambda kennung: time.time() - 3600 if gesperrt["an"] and kennung == "gitea" else 0.0)
|
||
homelab.ausfuehrer.weiter.clear()
|
||
assert sammellauf.starten()["ok"]
|
||
gesperrt["an"] = True # das Update-Skript von Gitea ändert sich mitten im Lauf
|
||
homelab.ausfuehrer.weiter.set()
|
||
sl = _auf_ende()
|
||
assert sl["status"] == "fertig"
|
||
assert [s["status"] for s in sl["schritte"]] == ["eingespielt", "uebersprungen"] + ["eingespielt"] * 4
|
||
assert sl["schritte"][1]["text"].startswith("Das Update-Skript wurde am ")
|
||
assert sl["text"].startswith("5 Updates eingespielt, alle Prüfungen grün. Nicht gelaufen: Gitea (Das Update-")
|
||
assert homelab.meldungen == [("Homelab: " + sl["text"], False)]
|
||
|
||
|
||
# --- Sperre ---------------------------------------------------------------------------------------------
|
||
|
||
def test_nur_ein_sammellauf_und_kein_einzelner_knopf_waehrenddessen(homelab):
|
||
homelab.ausfuehrer.weiter.clear() # der erste Schritt hängt am Snapshot
|
||
assert sammellauf.starten()["ok"]
|
||
assert sammellauf.starten() == {"ok": False, "detail": SPERRE}
|
||
assert updates.starten("ct-104", "app") == {"ok": False, "detail": SPERRE}
|
||
assert updates.starten("pve", "neustart") == {"ok": False, "detail": SPERRE}
|
||
assert updates.starten("vm-106", "docker") == {"ok": False, "detail": SPERRE}
|
||
homelab.ausfuehrer.weiter.set()
|
||
assert _auf_ende()["status"] == "fertig"
|
||
assert not sammellauf.laeuft()
|
||
|
||
|
||
def test_nicht_neben_einem_einzelnen_update(homelab):
|
||
homelab.ausfuehrer.weiter.clear()
|
||
lauf = updates.starten("ct-104", "app")
|
||
assert lauf["ok"]
|
||
assert sammellauf.starten() == {"ok": False, "detail": "Gerade läuft ein einzelnes Update (Gitea). „Alle "
|
||
"aktualisieren“ geht, sobald es fertig ist."}
|
||
homelab.ausfuehrer.weiter.set()
|
||
assert _auf_lauf(lauf["lauf"])["ergebnis"] == "eingespielt"
|
||
assert homelab.meldungen == [("Gitea: eingespielt, Prüfung grün.", False)] # allein: wie bisher gemeldet
|
||
assert sammellauf.aktueller() is None
|
||
|
||
|
||
def test_ohne_ausfuehrer_im_nur_lesen_modus_und_ohne_updates(homelab, monkeypatch):
|
||
monkeypatch.setattr(kanal, "zuletzt", lambda: time.time() - 3600)
|
||
assert sammellauf.starten() == {"ok": False, "detail": "Der Ausführer auf dem Proxmox-Host ist nicht verbunden. "
|
||
"Ohne ihn geht kein Update."}
|
||
monkeypatch.setattr(kanal, "zuletzt", lambda: time.time())
|
||
nur_lesen = _bericht()
|
||
nur_lesen["host"]["nur_lesen"] = True
|
||
kanal.bericht_speichern(nur_lesen)
|
||
assert "Nur-Lesen-Modus" in sammellauf.starten()["detail"]
|
||
(homelab.daten / "pve-bericht.json").unlink()
|
||
assert sammellauf.starten() == {"ok": False, "detail": "Gerade gibt es kein Update mit Knopf, also nichts zu tun."}
|
||
assert sammellauf.aktueller() is None and homelab.ausfuehrer.protokoll == []
|
||
|
||
|
||
# --- Neustart und Absturz -------------------------------------------------------------------------------
|
||
|
||
def _offener_sammellauf(gitea_lauf: dict) -> None:
|
||
"""So stand es auf der Platte, als der Homelab-Teil mitten im Gitea-Schritt neu startete."""
|
||
jetzt = time.time()
|
||
updates._merken({"id": "l-netbird", "ziel": "ct-102", "baustein": "app", "name": "NetBird", "status": "fertig",
|
||
"ergebnis": "eingespielt", "text": "NetBird: eingespielt, Prüfung grün.", "start": jetzt - 300,
|
||
"ende": jetzt - 200, "schritte": [], "sammellauf": "s1", "sammellauf_start": jetzt - 400})
|
||
updates._merken({"id": "l-gitea", "ziel": "ct-104", "baustein": "app", "name": "Gitea", "start": jetzt - 100,
|
||
"ende": None, "ergebnis": None, "text": None, "schritte": ["update …"], "sammellauf": "s1",
|
||
"sammellauf_start": jetzt - 400, **gitea_lauf})
|
||
schritt = {"baustein": "app", "lauf": None, "text": None}
|
||
sammellauf._merken({"id": "s1", "status": "laeuft", "start": jetzt - 400, "ende": None, "aktuell": 1,
|
||
"text": None, "meldung": None, "dringend": False, "schritte": [
|
||
{**schritt, "ziel": "ct-102", "name": "NetBird", "was": "0.77.1 → 0.79.0",
|
||
"status": "eingespielt", "lauf": "l-netbird", "text": "NetBird: eingespielt, Prüfung grün."},
|
||
{**schritt, "ziel": "ct-104", "name": "Gitea", "was": "1.27.2 → 1.27.3", "status": "laeuft",
|
||
"lauf": "l-gitea"},
|
||
{**schritt, "ziel": "pve", "name": "Proxmox-Host", "baustein": "pakete", "was": "5 Pakete",
|
||
"status": "wartet"}]})
|
||
|
||
|
||
def _auf_meldungen(meldungen: list, anzahl: int) -> list:
|
||
ende = time.time() + 10
|
||
while len(meldungen) < anzahl and time.time() < ende:
|
||
time.sleep(0.02)
|
||
return meldungen
|
||
|
||
|
||
def test_neustart_mitten_im_lauf_bricht_ab(homelab):
|
||
_offener_sammellauf({"status": "laeuft"})
|
||
assert updates.starten("ct-104", "app") == {"ok": False, "detail": SPERRE}
|
||
updates.unterbrochene_abschliessen() # so ruft es der Start der App (Lebenszyklus in app.py)
|
||
sl = sammellauf.aktueller()
|
||
assert (sl["status"], sl["text"], sl["aktuell"]) == ("abgebrochen", "unterbrochen (Neustart)", None)
|
||
assert sl["ende"] is not None
|
||
assert [(s["status"], s["text"]) for s in sl["schritte"]] == [
|
||
("eingespielt", "NetBird: eingespielt, Prüfung grün."), ("fehler", "unterbrochen (Neustart)"),
|
||
("uebersprungen", "Nicht mehr gelaufen: „Alle aktualisieren“ wurde abgebrochen.")]
|
||
assert updates.lauf_lesen("l-gitea")["status"] == "unterbrochen"
|
||
assert _auf_meldungen(homelab.meldungen, 1) == [(
|
||
("Homelab: „Alle aktualisieren“ wurde unterbrochen: Der Homelab-Teil ist mitten im Lauf neu gestartet (bei "
|
||
"Gitea). Vorher eingespielt: NetBird. Nicht mehr gelaufen: Proxmox-Host. Bitte den Stand in der Übersicht "
|
||
"prüfen."), True)]
|
||
assert not sammellauf.laeuft() # der einzelne Knopf geht wieder
|
||
|
||
|
||
def test_neustart_nach_fertigem_schritt_zaehlt_dessen_ergebnis(homelab):
|
||
_offener_sammellauf({"status": "fertig", "ergebnis": "eingespielt", "text": "Gitea: eingespielt, Prüfung grün.",
|
||
"ende": time.time()})
|
||
updates.unterbrochene_abschliessen()
|
||
sl = sammellauf.aktueller()
|
||
assert [s["status"] for s in sl["schritte"]] == ["eingespielt", "eingespielt", "uebersprungen"]
|
||
assert sl["status"] == "abgebrochen" and sl["text"] == "unterbrochen (Neustart)"
|
||
assert "Vorher eingespielt: NetBird und Gitea." in _auf_meldungen(homelab.meldungen, 1)[0][0]
|
||
|
||
|
||
def test_absturz_im_ablauf_gibt_die_sperre_frei(homelab, monkeypatch):
|
||
def kaputt(sl, schritt):
|
||
raise KeyError("kaputt")
|
||
monkeypatch.setattr(sammellauf, "_schritt", kaputt)
|
||
assert sammellauf.starten()["ok"]
|
||
sl = _auf_ende()
|
||
assert sl["status"] == "abgebrochen" and sl["text"] == "Interner Fehler: KeyError."
|
||
assert [s["status"] for s in sl["schritte"]] == ["uebersprungen"] * 6
|
||
assert homelab.meldungen[-1][1] is True and not sammellauf.laeuft()
|
||
|
||
|
||
# --- Schnittstellen -------------------------------------------------------------------------------------
|
||
|
||
def _client() -> TestClient:
|
||
from routers import homelab as homelab_router
|
||
app = FastAPI()
|
||
app.include_router(homelab_router.router)
|
||
return TestClient(app)
|
||
|
||
|
||
def test_schnittstellen(homelab):
|
||
c = _client()
|
||
assert c.get("/api/homelab/sammellauf").json() == {"sammellauf": None}
|
||
plan = c.get("/api/homelab/alle/plan").json()
|
||
assert [(s["ziel"], s["baustein"]) for s in plan["schritte"]] == ALLE and plan["hinweis"]
|
||
homelab.ausfuehrer.weiter.clear()
|
||
antwort = c.post("/api/homelab/alle")
|
||
assert antwort.status_code == 200 and antwort.json()["ok"] is True
|
||
zweite = c.post("/api/homelab/alle")
|
||
assert zweite.status_code == 200 and zweite.json() == {"ok": False, "detail": SPERRE}
|
||
einzeln = c.post("/api/homelab/ziele/ct-104/update")
|
||
assert einzeln.status_code == 409 and einzeln.json()["detail"] == SPERRE
|
||
ende = time.time() + 10 # der Faden setzt gleich den ersten Schritt auf „läuft“
|
||
while (laufend := c.get("/api/homelab/sammellauf").json()["sammellauf"])["aktuell"] is None and time.time() < ende:
|
||
time.sleep(0.02)
|
||
assert laufend["status"] == "laeuft" and laufend["aktuell"] == 0 and laufend["schritte"][0]["status"] == "laeuft"
|
||
homelab.ausfuehrer.weiter.set()
|
||
_auf_ende()
|
||
fertig = c.get("/api/homelab/sammellauf").json()["sammellauf"]
|
||
assert fertig["status"] == "fertig" and fertig["id"] == antwort.json()["sammellauf"]["id"]
|
||
assert "meldung" not in fertig and "dringend" not in fertig
|
||
|
||
|
||
def test_quittiert_verschwindet_von_der_update_seite(homelab):
|
||
"""25.09.2026: Das fertige „Alle aktualisieren“ ließ sich nicht wegklicken."""
|
||
antwort = sammellauf.starten()
|
||
assert sammellauf.quittieren(antwort["sammellauf"]["id"]) is False # läuft noch
|
||
sl = _auf_ende()
|
||
assert sammellauf.aktueller()["id"] == sl["id"]
|
||
assert sammellauf.quittieren(sl["id"]) is True
|
||
assert sammellauf.aktueller() is None
|
||
assert sammellauf.liste()[0]["quittiert"] is True # das Protokoll behält ihn
|