HitonabiandClaude Opus 5.5
afa3a476b5
Der Ausführer meldet jetzt, welche Gäste kein Sicherungsauftrag erfasst (wie Proxmox unter Rechenzentrum → Backup), je Gast die jüngste Prüfung durch den PBS samt beschädigter Sicherungen, die Schätzung des PBS, wann sein Datastore voll ist, und öffnet einmal die Woche die jüngste Sicherung jedes Gasts probeweise (nur lesend: Konfiguration, bei Containern der Dateibaum). Der Wächter meldet: ohne Auftrag gelb (sofort), beschädigt rot, seit 15 Tagen nicht geprüft gelb, Probe gescheitert gelb, Sicherungsziel laut PBS bald voll. Die Speicherkarte zeigt „Sicherungen je Gerät“ mit Knopf „Jetzt prüfen“. Der Ausführer braucht ausfuehrer-einrichten.sh. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
220 lines
13 KiB
Python
220 lines
13 KiB
Python
"""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()
|