"""PBS: Sicherung statt Snapshot (24.09.2026). Der Ausführer über Fakes (die Speicher und CT 105 so, wie sie am 24.09. auf dem Proxmox-Host standen: pvesm status, storage.cfg, pct config 105), der Ablauf im Homelab-Teil mit einem nachgespielten Ausführer.""" import importlib.util import json import threading import time from pathlib import Path import pytest 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" GIB = 1024 ** 3 # /nodes/pve/storage am 24.09.2026 (Zahlen aus pvesm status, KiB → Bytes). SPEICHER = [ {"storage": "local", "type": "dir", "content": "vztmpl,backup,iso", "active": 1, "enabled": 1, "shared": 0, "avail": 46835092 * 1024, "used": 46613140 * 1024, "total": 98497780 * 1024}, {"storage": "local-lvm", "type": "lvmthin", "content": "rootdir,images", "active": 1, "enabled": 1, "shared": 0, "avail": 186793893 * 1024}, {"storage": "pbs-qnap", "type": "pbs", "content": "backup", "active": 1, "enabled": 1, "shared": 1, "avail": 2282130304 * 1024}, ] KONFIG_105 = {"rootfs": "local-lvm:vm-105-disk-0,size=10G", "mp0": "/mnt/qnap-backup/pbs-datastore,mp=/mnt/datastore", "unprivileged": 1, "onboot": 1, "ostype": "debian", "tags": "backup;community-script"} ALT = "local:backup/vzdump-lxc-105-2026_09_20-03_00_00.tar.zst" NOTIZ = "mc2-sicherung: vor einem Update durch den Homelab Orchestrator (proxmox-backup-server)" def _modul(): spec = importlib.util.spec_from_file_location("ausfuehrer", AUSFUEHRER) modul = importlib.util.module_from_spec(spec) spec.loader.exec_module(modul) return modul # --- Reine Auswertungen im Ausführer ---------------------------------------------------------- def test_speicher_wahl_nimmt_local_und_nie_den_pbs(): a = _modul() s, fehlt = a.sicherungsspeicher(SPEICHER) assert s["storage"] == "local" and fehlt is None # Ohne lokalen Speicher mit Sicherungen: klare Ablehnung, der PBS ist keine Ausweichlösung. s, fehlt = a.sicherungsspeicher([x for x in SPEICHER if x["storage"] != "local"]) assert s is None and "Kein lokaler Speicher" in fehlt and "VZDump-Sicherung" in fehlt assert a.sicherungsspeicher([])[1].startswith("Die Speicher des Proxmox-Hosts") # Ein lokaler Speicher, der nur als „geteilt“ eingetragen ist (Netzfreigabe), zählt nicht als lokal. geteilt = [{**SPEICHER[0], "shared": 1}] + SPEICHER[1:] assert a.sicherungsspeicher(geteilt)[0] is None def test_speicher_wunsch_aus_der_konfiguration(): a = _modul() assert a.sicherungsspeicher(SPEICHER, "local")[0]["storage"] == "local" s, fehlt = a.sicherungsspeicher(SPEICHER, "pbs-qnap") assert s is None and "nie" in fehlt and "selbst sichern" in fehlt assert "keine Sicherungen" in a.sicherungsspeicher(SPEICHER, "local-lvm")[1] assert "gibt es nicht" in a.sicherungsspeicher(SPEICHER, "usb")[1] aus = [{**SPEICHER[0], "active": 0}] assert "nicht verfügbar" in a.sicherungsspeicher(aus, "local")[1] def test_platzbedarf_ohne_bind_mount(): a = _modul() # Läuft der Gast: die belegte rootfs. Der Bind-Mount (Datenspeicher des PBS) zählt nicht. assert a.sicherung_bedarf(KONFIG_105, 3 * GIB) == 3 * GIB # Läuft er nicht: die volle Größe der rootfs. assert a.sicherung_bedarf(KONFIG_105, None) == 10 * GIB # Weitere Volumes zählen nur mit backup=1. mit = {**KONFIG_105, "mp1": "local-lvm:vm-105-disk-1,mp=/daten,backup=1,size=8G", "mp2": "local-lvm:vm-105-disk-2,mp=/tmp2,size=4G"} assert a.sicherung_bedarf(mit, 3 * GIB) == 11 * GIB assert a.sicherung_bedarf({}, None) is None def test_platzpruefung_frei_groesser_als_belegt_mal_1_2(): a = _modul() stand = a.sicherung_stand(SPEICHER) assert stand == {"speicher": "local", "art": "dir", "frei": 46835092 * 1024} assert a.sicherung_pruefen(stand, 3 * GIB) == (True, None) eng = {"speicher": "local", "frei": int(3.5 * GIB)} ok, grund = a.sicherung_pruefen(eng, 3 * GIB) # 3,5 GB frei < 3 GB × 1,2 assert not ok and grund == "Zu wenig Platz auf „local“: frei 3,5 GB, nötig mehr als 3,6 GB (belegt 3,0 GB × 1,2)." assert a.sicherung_pruefen({"speicher": "local", "frei": int(3.7 * GIB)}, 3 * GIB)[0] is True assert a.sicherung_pruefen({"speicher": None, "fehler": "Kein lokaler Speicher …"}, 3 * GIB) == ( False, "Kein lokaler Speicher …") def test_modus_snapshot_wo_die_rootfs_es_kann(): a = _modul() arten = {s["storage"]: s["type"] for s in SPEICHER} assert a.sicherung_modus(KONFIG_105, arten) == "snapshot" assert a.sicherung_modus({"rootfs": "local:105/vm-105-disk-0.raw,size=8G"}, arten) == "stop" def test_nur_eigene_sicherungen_zaehlen(): a = _modul() eintraege = [ {"volid": "local:backup/vzdump-lxc-105-2026_09_24-22_00_00.tar.zst", "ctime": 20, "notes": NOTIZ}, {"volid": ALT, "ctime": 10, "notes": NOTIZ}, {"volid": "local:backup/vzdump-lxc-105-2026_09_01-03_00_00.tar.zst", "ctime": 5, "notes": "von Hand"}, {"volid": "local:backup/vzdump-lxc-104-2026_09_24-22_00_00.tar.zst", "ctime": 30, "notes": NOTIZ}, {"volid": "pbs-qnap:backup/ct/105/2026-09-24T20:00:00Z", "ctime": 40, "notes": NOTIZ}, ] assert [e["volid"] for e in a.eigene_sicherungen(eintraege, 105)] == [ ALT, "local:backup/vzdump-lxc-105-2026_09_24-22_00_00.tar.zst"] assert len(a.eigene_sicherungen(eintraege)) == 3 assert a.eigene_sicherungen(None) == [] # --- Aktionen des Ausführers über Fakes ------------------------------------------------------- class Host: """Der Proxmox-Host als Fake: beantwortet pvesh und führt „Befehle“ nur ins Protokoll.""" def __init__(self, a, speicher=SPEICHER, belegt=3 * GIB): self.befehle: list[list[str]] = [] self.inhalt = [{"volid": ALT, "vmid": 105, "ctime": 1790000000, "notes": NOTIZ}, {"volid": "local:backup/vzdump-lxc-105-2026_09_01-03_00_00.tar.zst", "vmid": 105, "ctime": 1788000000, "notes": "von Hand"}] self.status = "running" self.antworten = { "/cluster/resources": [{"vmid": 105, "type": "lxc", "tags": "backup;community-script"}, {"vmid": 106, "type": "qemu", "tags": ""}], "/nodes/pve/storage": speicher, "/nodes/pve/lxc/105/config": KONFIG_105, "/nodes/pve/lxc/105/status/current": lambda: {"status": self.status, "disk": belegt}, "/nodes/pve/storage/local/content": lambda: self.inhalt, } a.NODE = "pve" a._pvesh = self.pvesh a._laufen = self.laufen def pvesh(self, pfad, *parameter): if pfad not in self.antworten: raise RuntimeError(f"pvesh {pfad}: gibt es im Fake nicht") wert = self.antworten[pfad] return wert() if callable(wert) else wert def laufen(self, befehl, zeitlimit=60, sanft=False): self.befehle.append(list(befehl)) if befehl[0] == "vzdump": self.sanft_vzdump = sanft self.inhalt.append({"volid": "local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst", "vmid": 105, "ctime": int(time.time()), "notes": NOTIZ}) return 0, "INFO: Backup job finished successfully" if befehl[:2] == ["pct", "shutdown"]: self.status = "stopped" return 0, "ok" def test_sichern_vzdump_auf_local_mit_notiz(monkeypatch): a = _modul() host = Host(a) antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}}) assert antwort["code"] == 0 assert antwort["text"].endswith("\nsicherung=local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst") befehl = host.befehle[-1] assert befehl[:2] == ["vzdump", "105"] and host.sanft_vzdump is True optionen = dict(zip(befehl[2::2], befehl[3::2], strict=True)) assert optionen["--storage"] == "local" and optionen["--mode"] == "snapshot" assert optionen["--remove"] == "0" and optionen["--compress"] == "zstd" assert optionen["--notes-template"].startswith("mc2-sicherung") def test_sichern_lehnt_ab_ohne_speicher_und_ohne_platz(monkeypatch): a = _modul() host = Host(a, speicher=[s for s in SPEICHER if s["storage"] != "local"]) antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}}) assert antwort["code"] == 2 and "Kein lokaler Speicher" in antwort["text"] and host.befehle == [] host = Host(a, speicher=[{**SPEICHER[0], "avail": 2 * GIB}] + SPEICHER[1:], belegt=3 * GIB) antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}}) assert antwort["code"] == 2 and "Zu wenig Platz" in antwort["text"] and host.befehle == [] # Wunsch in der Konfiguration zeigt auf den PBS: nie. host = Host(a) monkeypatch.setattr(a, "SICHERUNG_SPEICHER", "pbs-qnap") antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}}) assert antwort["code"] == 2 and "Proxmox-Backup-Server" in antwort["text"] and host.befehle == [] # Ein Gast ohne erlaubtes Etikett bleibt unangetastet. assert "kein erlaubtes Etikett" in a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 106}})["text"] def test_sichern_ohne_kennzeichnung_ist_rot(): a = _modul() host = Host(a) def ohne_notiz(befehl, zeitlimit=60, sanft=False): host.befehle.append(befehl) host.inhalt.append({"volid": "local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst", "vmid": 105, "ctime": int(time.time()), "notes": ""}) return 0, "INFO: fertig" a._laufen = ohne_notiz antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}}) assert antwort["code"] == 1 and "mc2-sicherung" in antwort["text"] def test_sicherung_zurueck_stoppt_spielt_zurueck_und_startet(): a = _modul() host = Host(a) antwort = a.ausfuehren({"aktion": "sicherung_zurueck", "parameter": {"vmid": 105, "sicherung": ALT}}) assert antwort["code"] == 0 assert host.befehle == [ ["pct", "shutdown", "105", "--timeout", "120", "--forceStop", "1"], ["pct", "restore", "105", ALT, "--force", "1", "--storage", "local-lvm"], # der bisherige rootfs-Speicher ["pct", "start", "105"]] def test_zurueckgespielt_und_geloescht_wird_nur_eigenes(): a = _modul() host = Host(a) fremd = "local:backup/vzdump-lxc-105-2026_09_01-03_00_00.tar.zst" # von Hand, ohne mc2-Notiz for aktion in ("sicherung_zurueck", "sicherung_loeschen"): for volid in (fremd, "local:backup/vzdump-lxc-104-2026_09_20-03_00_00.tar.zst", "pbs-qnap:backup/ct/105/2026-09-24T20:00:00Z", "/var/lib/vz/dump/x.tar.zst; reboot", None): antwort = a.ausfuehren({"aktion": aktion, "parameter": {"vmid": 105, "sicherung": volid}}) assert antwort["code"] == 2, (aktion, volid) assert host.befehle == [] assert a.ausfuehren({"aktion": "sicherung_loeschen", "parameter": {"vmid": 105, "sicherung": ALT}})["code"] == 0 assert host.befehle == [["pvesm", "free", ALT]] def test_laufen_mit_zeitlimit_und_sanftem_abbruch(): import sys a = _modul() assert a._laufen([sys.executable, "-c", "print('hallo')"]) == (0, "hallo") beginn = time.time() code, text = a._laufen([sys.executable, "-c", "import time; time.sleep(30)"], zeitlimit=1, sanft=True) assert (code, text) == (124, "Zeitlimit (1 s) überschritten") and time.time() - beginn < 20 assert a._laufen(["gibt-es-nicht-mc2"])[0] == 127 def test_unerwarteter_fehler_wird_beantwortet(): """Sonst bliebe der Auftrag offen, und der Homelab-Teil wartete bis zu seinem Zeitlimit.""" a = _modul() Host(a) a.AKTIONEN["sichern"] = lambda p: {}["kaputt"] antwort = a.ausfuehren({"aktion": "sichern", "parameter": {"vmid": 105}}) assert antwort["code"] == 1 and "KeyError" in antwort["text"] def test_bericht_sagt_ob_eine_sicherung_geht(): a = _modul() host = Host(a) host.antworten.update({ "/cluster/resources": [{"vmid": 105, "type": "lxc", "node": "pve", "name": "proxmox-backup-server", "status": "running", "tags": "backup;community-script", "disk": 3 * GIB}], "/nodes/pve/lxc/105/snapshot": [{"name": "current"}], "/version": {"version": "9.2.11"}, "/nodes/pve/apt/update": [], }) bericht = a.bericht() assert bericht["host"]["sicherung"] == {"speicher": "local", "art": "dir", "frei": 46835092 * 1024} assert bericht["host"]["nur_lesen"] is False # die Pflege legt sonst keine Aufträge an pbs = bericht["gaeste"][0] assert pbs["snapshot_moeglich"] is False and "Bind-Mount" in pbs["snapshot_grund"] assert (pbs["sicherung_moeglich"], pbs["sicherung_grund"], pbs["sicherung_speicher"]) == (True, None, "local") assert pbs["sicherungen"] == [ALT] # die fremde Sicherung von Hand zählt nicht host.antworten["/nodes/pve/storage"] = SPEICHER[1:] # kein lokaler Speicher mit Sicherungen pbs = a.bericht()["gaeste"][0] assert pbs["sicherung_moeglich"] is False and "Kein lokaler Speicher" in pbs["sicherung_grund"] # --- Der Ablauf im Homelab-Teil --------------------------------------------------------------- @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}) monkeypatch.setattr(inventar, "neueste_version", lambda repo: None) monkeypatch.setattr(inventar, "erreichbar", lambda url: True) monkeypatch.setattr(inventar.karenz, "skript_geaendert", lambda kennung: 0.0) yield tmp_path einstellungen_mod.einstellungen.cache_clear() def _mit_sicherung(bericht: dict, **pbs) -> dict: neu = json.loads(json.dumps(bericht)) gast = next(g for g in neu["gaeste"] if g["vmid"] == 105) gast.update(pbs) gast["os_updates"]["listen_stand"] = int(time.time()) # frische Listen: 58 Pakete bleiben „neu“ return neu BERICHT_PBS = _mit_sicherung(BERICHT, sicherung_moeglich=True, sicherung_grund=None, sicherung_speicher="local", sicherungen=[ALT]) NEU = "local:backup/vzdump-lxc-105-2026_09_24-22_10_00.tar.zst" def _ausfuehrer(stopp: threading.Event, protokoll: list, fehlschlag: dict) -> None: """Nachgespielter Ausführer; fehlschlag: Aktion → (Code, Text).""" while not stopp.is_set(): auftrag = kanal.naechster() if not auftrag: time.sleep(0.02) continue protokoll.append((auftrag["aktion"], auftrag["parameter"])) code, text = fehlschlag.get(auftrag["aktion"], (0, "")) if auftrag["aktion"] == "sichern" and code == 0: text = f"INFO: fertig\nsicherung={NEU}" elif auftrag["aktion"] == "bericht": text = json.dumps(_mit_sicherung(BERICHT_PBS, sicherungen=[ALT, NEU])) kanal.ergebnis(auftrag["id"], code, text) @pytest.fixture def lauf(daten, monkeypatch): monkeypatch.setattr(updates, "WARTEN_NACH_UPDATE_S", 0) echt = kanal.warten monkeypatch.setattr(kanal, "warten", lambda aid, zeitlimit_s, takt_s=2.0: echt(aid, min(zeitlimit_s, 10), 0.02)) meldungen: list[tuple[str, bool]] = [] monkeypatch.setattr(updates, "_melden", lambda text, dringend=False: meldungen.append((text, dringend))) monkeypatch.setattr(updates.update_verlauf, "eintragen", lambda *a: None) kanal.bericht_speichern(BERICHT_PBS) stopp, protokoll, fehlschlag = threading.Event(), [], {} def starten(ziel: str, baustein: str) -> dict: threading.Thread(target=_ausfuehrer, args=(stopp, protokoll, fehlschlag), daemon=True).start() antwort = updates.starten(ziel, baustein) assert antwort["ok"], antwort ende = time.time() + 20 while time.time() < ende: x = next((x for x in updates.laeufe() if x["id"] == antwort["lauf"]), {}) if x.get("status") == "fertig": return x time.sleep(0.05) raise AssertionError("Lauf wurde nicht fertig") yield starten, protokoll, meldungen, fehlschlag stopp.set() def test_pbs_gruen_mit_sicherung_und_alte_sicherung_weg(lauf): starten, protokoll, meldungen, _ = lauf x = starten("ct-105", "os") assert x["ergebnis"] == "eingespielt" assert protokoll == [("sichern", {"vmid": 105}), ("os_update", {"vmid": 105}), ("bericht", {}), ("sicherung_loeschen", {"vmid": 105, "sicherung": ALT})] # die neue bleibt assert meldungen == [("Proxmox Backup Server: eingespielt, Prüfung grün.", False)] def test_pbs_rot_spielt_die_sicherung_zurueck(lauf, monkeypatch): starten, protokoll, meldungen, _ = lauf monkeypatch.setattr(inventar, "erreichbar", lambda url: False) x = starten("ct-105", "os") assert x["ergebnis"] == "zurueckgerollt" assert [p[0] for p in protokoll] == ["sichern", "os_update", "bericht", "sicherung_zurueck"] assert protokoll[-1][1] == {"vmid": 105, "sicherung": NEU} text, dringend = meldungen[0] assert dringend and "Sicherung von vorher zurückgespielt (vzdump-lxc-105-2026_09_24-22_10_00.tar.zst)" in text def test_pbs_rot_und_auch_das_zurueckspielen_scheitert(lauf, monkeypatch): starten, _, meldungen, fehlschlag = lauf monkeypatch.setattr(inventar, "erreichbar", lambda url: False) fehlschlag["sicherung_zurueck"] = (1, "restore failed") x = starten("ct-105", "os") assert x["ergebnis"] == "fehler" text, dringend = meldungen[0] assert dringend and "Auch der Rückweg scheiterte" in text and f"Die Sicherung {NEU} liegt weiter" in text def test_ohne_sicherung_beginnt_das_update_nicht(lauf): starten, protokoll, meldungen, fehlschlag = lauf fehlschlag["sichern"] = (2, "Abgelehnt: Zu wenig Platz auf „local“: frei 2,0 GB, nötig mehr als 3,6 GB.") x = starten("ct-105", "os") assert x["ergebnis"] == "fehler" and [p[0] for p in protokoll] == ["sichern"] text, dringend = meldungen[0] assert not dringend and "Update nicht begonnen" in text and "Zu wenig Platz" in text assert "nichts geändert" in text and ".." not in text def test_gescheiterter_snapshot_heisst_update_nicht_begonnen(lauf, monkeypatch): """Vorher hieß das „Update gescheitert … kein Rückweg“ (dringend), obwohl das Update nie lief.""" starten, protokoll, meldungen, fehlschlag = lauf kanal.bericht_speichern(BERICHT) monkeypatch.setattr(inventar, "neueste_version", lambda repo: { "tag": "v1.27.3", "tag_roh": "v1.27.3", "version": "1.27.3", "datum": "2026-09-20"}) fehlschlag["snapshot"] = (1, "snapshot feature is not available") x = starten("ct-104", "app") assert x["ergebnis"] == "fehler" and [p[0] for p in protokoll] == ["snapshot"] erwartet = ("Gitea: Update nicht begonnen — snapshot gescheitert: snapshot feature is not available. " "Am Gerät wurde nichts geändert.") assert meldungen == [(erwartet, False)] def test_weder_snapshot_noch_sicherung_wie_bisher(lauf, monkeypatch): starten, protokoll, meldungen, _ = lauf kanal.bericht_speichern(_mit_sicherung(BERICHT, sicherung_moeglich=False, sicherung_grund="Kein lokaler Speicher des Proxmox-Hosts …")) monkeypatch.setattr(inventar, "erreichbar", lambda url: False) x = starten("ct-105", "os") assert [p[0] for p in protokoll] == ["os_update", "bericht"] and x["ergebnis"] == "fehler" assert "weder Snapshot noch Sicherung" in meldungen[0][0] # --- Was die Übersicht dazu sagt ---------------------------------------------------------------- def _pbs(bericht: dict) -> dict: kanal.bericht_speichern(bericht) return {z["id"]: z for z in inventar.ziele()["ziele"]}["ct-105"] def test_rueckweg_und_rueckfrage_mit_sicherung(daten): pbs = _pbs(BERICHT_PBS) assert pbs["rueckweg"].startswith("Sicherung auf „local“ vor jedem Update (Bind-Mount mp0") os_baustein = {b["id"]: b for b in pbs["bausteine"]}["os"] assert os_baustein["aktion"]["frage"] == ("58 Pakete im Gast jetzt einspielen? Vorher wird eine Sicherung " "angelegt; ist die Prüfung danach rot, wird sie zurückgespielt.") def test_ohne_sicherung_sagt_die_rueckfrage_warum(daten): pbs = _pbs(_mit_sicherung(BERICHT, sicherung_moeglich=False, sicherung_grund="Zu wenig Platz auf „local“: frei 2,0 GB.")) assert "Auch keine Sicherung: Zu wenig Platz" in pbs["rueckweg"] frage = {b["id"]: b for b in pbs["bausteine"]}["os"]["aktion"]["frage"] assert "keinen automatischen Rückweg" in frage and "Eine Sicherung geht auch nicht: Zu wenig Platz" in frage # Der alte Bericht (Ausführer noch ohne Sicherungen) bleibt wie bisher. assert _pbs(BERICHT)["rueckweg"] == "Bind-Mount mp0 (/mnt/qnap-backup/pbs-datastore) — nur Backup möglich" def test_waechter_laesst_gaeste_im_update_in_ruhe(daten): from services import waechter kanal.bericht_speichern(BERICHT_PBS) updates._merken({"id": "x1", "ziel": "ct-105", "baustein": "os", "name": "PBS", "status": "laeuft", "start": time.time(), "ende": None, "ergebnis": None, "text": None, "schritte": []}) ids = [e["id"] for e in inventar.erreichbarkeit()] assert "ct-105" not in ids and "ct-104" in ids assert all(b.id != "gast:ct-105" for b in waechter.pruefe_gaeste())