"""Sicherungs-Abdeckung (Ausbauplan Welle 1, 25.09.2026): Gäste ohne Auftrag, die Prüfung durch den PBS, seine Schätzung, wann er voll ist, und die Probe-Wiederherstellung — im Ausführer (reine Auswertungen mit echten Zeilen vom Proxmox-PC, gelesen am 25.09.) und im Homelab-Teil (Karte, Hinweise, Knopf).""" import importlib.util import json from pathlib import Path import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from services.homelab import kanal, speicher AUSFUEHRER = Path(__file__).resolve().parents[2] / "deploy" / "homelab" / "ausfuehrer.py" JETZT = 1_790_340_000.0 # 25.09.2026 ~14:40 UPID_PBS = "UPID:proxmox-backup-server:0000620D:02F88CBE:00000126:6AAF5A40:verificationjob:qnap\\x3avj\\x2dqnap:root@pam:" GEPRUEFT = 0x6AAF5A40 # 20.09.2026, der wöchentliche Prüfauftrag „vj-qnap“ (So 06:00) @pytest.fixture(scope="module") def a(): spec = importlib.util.spec_from_file_location("ausfuehrer_abdeckung", AUSFUEHRER) modul = importlib.util.module_from_spec(spec) spec.loader.exec_module(modul) return modul # Echte Einträge von pvesh get /nodes/pve/storage/pbs-qnap/content --content backup (gekürzt) INHALT = [ {"vmid": 102, "ctime": 1789864223, "format": "pbs-ct", "size": 843_000_000, "volid": "pbs-qnap:backup/ct/102/2026-09-20T00:30:23Z", "verification": {"state": "ok", "upid": UPID_PBS}}, {"vmid": 102, "ctime": 1790296227, "format": "pbs-ct", "size": 843_394_679, "volid": "pbs-qnap:backup/ct/102/2026-09-25T00:30:27Z"}, {"vmid": 106, "ctime": 1789864278, "format": "pbs-vm", "size": 161_061_274_375, "volid": "pbs-qnap:backup/vm/106/2026-09-20T00:31:18Z", "verification": {"state": "failed", "upid": UPID_PBS}}, {"vmid": 106, "ctime": 1790296281, "format": "pbs-vm", "size": 161_061_274_375, "volid": "pbs-qnap:backup/vm/106/2026-09-25T00:31:21Z"}, {"vmid": "kaputt", "ctime": 1}, ] def test_pruefzeit_aus_der_upid(a): assert a.pruefzeit(UPID_PBS) == GEPRUEFT assert a.pruefzeit("UPID:pve:0000A1B2:0012C3D4:6AB5E4DC:vzdump::root@pam:") == 0x6AB5E4DC # Proxmox VE assert a.pruefzeit("kaputt") is None and a.pruefzeit(None) is None def test_details_je_gast(a): d = a.sicherungs_details(INHALT) assert set(d) == {"102", "106"} assert d["102"] == {"zeit": 1790296227, "aelteste": 1789864223, "volid": "pbs-qnap:backup/ct/102/2026-09-25T00:30:27Z", "format": "pbs-ct", "groesse": 843_394_679, "geprueft": {"zustand": "ok", "zeit": GEPRUEFT, "sicherung": 1789864223}, "defekt": []} assert d["106"]["geprueft"]["zustand"] == "failed" and d["106"]["defekt"] == [1789864278] def test_schaetzung_des_pbs(a): nutzung = [{"store": "qnap", "total": 3847503609856, "used": 1517685571584, "estimated-full-date": 1887231717}, {"store": "anderer", "estimated-full-date": 5}] assert a.pbs_schaetzung(nutzung, "qnap") == {"voll_am": 1887231717, "belegt": 1517685571584, "gesamt": 3847503609856} assert a.pbs_schaetzung([{"store": "qnap", "estimated-full-date": 0}], "qnap")["voll_am"] is None # wächst nicht assert a.pbs_schaetzung([{"store": "qnap", "error": "nicht erreichbar"}], "qnap") is None assert a.pbs_schaetzung("kaputt", "qnap") is None def test_probe_oeffnet_nur_lesend(a, monkeypatch): befehle: list[list[str]] = [] antworten = {"extractconfig": (0, "#Kommentar\narch: amd64\ncores: 2\n"), "file-restore/list": (0, json.dumps([{"text": "bin"}, {"text": "etc"}]))} def laufen(befehl, zeitlimit=60, sanft=False): befehle.append(befehl) return next(v for k, v in antworten.items() if any(k in teil for teil in befehl)) monkeypatch.setattr(a, "_laufen", laufen) monkeypatch.setattr(a, "NODE", "pve") assert a.probe_oeffnen("pbs-qnap:backup/ct/102/2026-09-25T00:30:27Z", "pbs-ct") is None assert befehle[0] == ["pvesm", "extractconfig", "pbs-qnap:backup/ct/102/2026-09-25T00:30:27Z"] assert befehle[1][:3] == ["pvesh", "get", "/nodes/pve/storage/pbs-qnap/file-restore/list"] assert "L3Jvb3QucHhhci5kaWR4" in befehle[1] # „/root.pxar.didx“ assert a.probe_oeffnen("pbs-qnap:backup/vm/106/2026-09-25T00:31:21Z", "pbs-vm") is None and len(befehle) == 3 antworten["file-restore/list"] = (0, json.dumps([{"text": "bin"}])) assert a.probe_oeffnen("x:backup/ct/1", "pbs-ct").startswith("Die Dateien der Sicherung ließen sich nicht auflisten") antworten["extractconfig"] = (255, "Error: unable to open archive\n") assert a.probe_oeffnen("x:backup/ct/1", "pbs-ct") == ( "Die Konfiguration ließ sich nicht aus der Sicherung lesen: Error: unable to open archive") def test_probe_nimmt_die_juengste_sicherung_je_gast(a): details = a.sicherungs_details(INHALT) ziele = {"pbs-qnap": {"aktiv": True, "details": details}, "weg": {"aktiv": False, "details": {"102": {**details["102"], "zeit": 1_800_000_000, "volid": "weg:x"}}}} geoeffnet: list[str] = [] stand = a.sicherungs_probe(ziele, oeffnen=lambda volid, art: geoeffnet.append(volid) or ( None if art == "pbs-ct" else "Die Konfiguration ließ sich nicht aus der Sicherung lesen")) assert geoeffnet == [details["102"]["volid"], details["106"]["volid"]] # das unerreichbare Ziel nicht assert stand["gaeste"]["102"] == {"speicher": "pbs-qnap", "sicherung": 1790296227, "ok": True, "fehler": None} assert stand["gaeste"]["106"]["ok"] is False def test_jetzt_pruefen_auch_im_nur_lesen_modus(a, monkeypatch): monkeypatch.setattr(a, "NUR_LESEN", True) monkeypatch.setattr(a, "bericht", lambda: {"host": {"sicherung_probe": {"zeit": 1, "gaeste": { "102": {"speicher": "pbs-qnap", "sicherung": 1, "ok": True, "fehler": None}}}}}) antwort = a.ausfuehren({"aktion": "sicherung_probe", "parameter": {}}) assert antwort["code"] == 0 and json.loads(antwort["text"])["gaeste"]["102"]["ok"] is True assert a._probe["naechste"] == 0.0 # der Bericht rechnete die Probe sofort (hier nachgestellt) # --- Homelab-Teil ------------------------------------------------------------------------------------------------ def _bericht(**host) -> dict: """Wie der echte Bericht vom 25.09.2026, drei Gäste; die Sicherungen kommen aus INHALT über den Ausführer.""" spec = importlib.util.spec_from_file_location("ausfuehrer_abdeckung_b", AUSFUEHRER) modul = importlib.util.module_from_spec(spec) spec.loader.exec_module(modul) details = modul.sicherungs_details(INHALT) basis = { "sicherung_jobs": [{"id": "j1", "zeitplan": "02:30", "speicher": "pbs-qnap", "an": True, "alle": False, "vmids": [102, 106], "ausgenommen": []}], "sicherungsziele": {"pbs-qnap": {"art": "pbs", "aktiv": True, "belegt": 1_517_685_571_584, "gesamt": 3_847_503_609_856, "details": details, "neueste": {k: v["zeit"] for k, v in details.items()}, "schaetzung": {"voll_am": 1887231717, "belegt": 1_517_685_571_584, "gesamt": 3_847_503_609_856}}}, "nicht_gesichert": [104], "sicherung_probe": {"zeit": 1790330000, "gaeste": { "102": {"speicher": "pbs-qnap", "sicherung": 1790296227, "ok": True, "fehler": None}, "106": {"speicher": "pbs-qnap", "sicherung": 1790296281, "ok": False, "fehler": "Die Konfiguration ließ sich nicht aus der Sicherung lesen: Error: timeout"}}}, } return {"host": {**basis, **host}, "gaeste": [ {"vmid": 102, "art": "lxc", "name": "netbird", "app": {"kennung": "netbird"}}, {"vmid": 104, "art": "lxc", "name": "gitea", "app": {"kennung": "gitea"}}, {"vmid": 106, "art": "qemu", "name": "arcane"}]} def test_abdeckung_je_geraet(): s = speicher.sicherungen(_bericht(), JETZT) zeilen = {g["ziel"]: g for g in s["geraete"]} assert [g["ziel"] for g in s["ohne_auftrag"]] == ["ct-104"] assert zeilen["ct-104"]["auftrag"] is False and zeilen["ct-104"]["zuletzt"] is None assert zeilen["ct-102"] == {"ziel": "ct-102", "name": zeilen["ct-102"]["name"], "vmid": 102, "auftrag": True, "zuletzt": 1790296227, "frisch": True, "geprueft": {"zustand": "ok", "zeit": GEPRUEFT, "sicherung": 1789864223}, "defekt": [], "probe": {"speicher": "pbs-qnap", "sicherung": 1790296227, "ok": True, "fehler": None, "zeit": 1790330000}} assert zeilen["vm-106"]["defekt"] == [1789864278] and zeilen["vm-106"]["probe"]["ok"] is False # Die letzte Prüfung von Arcane fand Beschädigtes: rot, auch wenn alles frisch gesichert ist. assert (s["stufe_frische"], s["stufe"]) == ("ok", "rot") assert s["pruefung"] == {"zuletzt": GEPRUEFT, "noetig": False} # älteste Sicherung erst 5,5 Tage alt assert s["probe"] == {"zeit": 1790330000} def test_hinweise_zur_abdeckung(monkeypatch): monkeypatch.setattr(speicher, "antwortet", lambda s, p: True) monkeypatch.setattr(speicher, "_bericht", lambda: _bericht()) befunde = {b.id: b for b in speicher.pruefe_speicher(JETZT)} assert "sicherungen" not in befunde # frisch ist alles ohne = befunde["sicherung:ohne-auftrag"] assert (ohne.stufe, ohne.sofort) == ("gelb", True) and ohne.titel == "Gitea ist in keinem Sicherungsauftrag" defekt = befunde["sicherung:defekt"] assert defekt.stufe == "rot" and "Arcane (Docker) (Sicherung vom 20.09.)" in defekt.text probe = befunde["sicherung:probe"] assert probe.stufe == "gelb" and probe.titel == "Probe-Wiederherstellung gescheitert: Arcane (Docker)" assert "sicherung:pruefung" not in befunde and "ziel:pbs-qnap" not in befunde # Drei Wochen später ohne neue Prüfung, und der PBS schätzt: in zehn Tagen voll. spaeter = JETZT + 21 * 86400 b = _bericht() b["host"]["sicherungsziele"]["pbs-qnap"]["schaetzung"]["voll_am"] = spaeter + 10 * 86400 monkeypatch.setattr(speicher, "_bericht", lambda: b) befunde = {x.id: x for x in speicher.pruefe_speicher(spaeter)} assert befunde["sicherung:pruefung"].titel == "Der PBS hat seit 26 Tagen keine Sicherung geprüft" ziel = befunde["ziel:pbs-qnap"] assert (ziel.stufe, ziel.titel) == ("gelb", "Sicherungsziel pbs-qnap in etwa 10 Tagen voll") def test_schaetzung_des_pbs_ist_die_vorhersage_des_ziels(monkeypatch): monkeypatch.setattr(speicher, "antwortet", lambda s, p: True) monkeypatch.setattr(speicher, "_bericht", lambda: _bericht()) (ziel,) = speicher.lage(JETZT)["sicherungen"]["ziele"] v = ziel["vorhersage"] assert v["quelle"] == "pbs" and v["tage"] == pytest.approx((1887231717 - JETZT) / 86400, abs=0.1) assert v["anteil"] == pytest.approx(39.45, abs=0.01) and v["grund"] is None b = _bericht() b["host"]["sicherungsziele"]["pbs-qnap"]["schaetzung"]["voll_am"] = None monkeypatch.setattr(speicher, "_bericht", lambda: b) assert speicher.lage(JETZT)["sicherungen"]["ziele"][0]["vorhersage"]["grund"] == "gleich" def test_aelterer_ausfuehrer_ohne_neue_angaben(): """Bis ausfuehrer-einrichten.sh gelaufen ist, fehlen nicht_gesichert, details und die Probe.""" b = _bericht() for schluessel in ("nicht_gesichert", "sicherung_probe"): b["host"].pop(schluessel) b["host"]["sicherungsziele"]["pbs-qnap"].pop("details") s = speicher.sicherungen(b, JETZT) assert [g["ziel"] for g in s["ohne_auftrag"]] == ["ct-104"] # selbst ausgerechnet aus den Aufträgen assert s["stufe"] == "gelb" and s["probe"] is None assert all(g["geprueft"] is None and g["probe"] is None for g in s["geraete"]) def test_knopf_jetzt_pruefen(tmp_path, monkeypatch): from kern import einstellungen as einstellungen_mod from routers import homelab monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path)) einstellungen_mod.einstellungen.cache_clear() try: app = FastAPI() app.include_router(homelab.router) client = TestClient(app) monkeypatch.setattr(kanal, "_kontakt", {"zuletzt": None}) antwort = client.post("/api/homelab/sicherungen/probe") assert antwort.status_code == 409 and "meldet sich gerade nicht" in antwort.json()["detail"] kanal.kontakt() assert client.post("/api/homelab/sicherungen/probe").json()["ok"] is True assert [a["aktion"] for a in kanal.liste()] == ["sicherung_probe"] assert client.post("/api/homelab/sicherungen/probe").json()["detail"] == "Die Probe läuft schon." finally: einstellungen_mod.einstellungen.cache_clear()