468 lines
26 KiB
Python
468 lines
26 KiB
Python
"""Seite Homelab → Snapshots (25.09.2026): Liste je Gerät, Aktionen als Läufe, der Ausführer mit Snapshot-Details.
|
|
|
|
Die Snapshots stammen aus dem echten Stand vom 25.09.2026 (pvesh get /nodes/pve/lxc/104/snapshot: nach „Alle
|
|
aktualisieren“ lag an jedem Container ein mc2-Snapshot), die Sicherung aus /nodes/pve/storage/local/content."""
|
|
|
|
import importlib.util
|
|
import json
|
|
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, protokoll, sammellauf, snapshots, 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"
|
|
UPDATE = "Vor einem Update durch den Homelab Orchestrator"
|
|
KNOPF = "Auf Knopfdruck im Homelab Orchestrator angelegt"
|
|
GITEA_SNAP = "mc2-20260925-112639" # snaptime 1790328400 = 25.09.2026 11:26:40
|
|
ALT = "local:backup/vzdump-lxc-105-2026_09_25-11_29_57.tar.zst"
|
|
NEU = "mc2-20260925-130512"
|
|
|
|
|
|
def _ausfuehrer_modul():
|
|
spec = importlib.util.spec_from_file_location("ausfuehrer_snapshots", AUSFUEHRER)
|
|
modul = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(modul)
|
|
return modul
|
|
|
|
|
|
# --- Ausführer: Snapshot-Details im Bericht, Anlass beim Anlegen -------------------------------------------------------
|
|
|
|
# pvesh get /nodes/pve/lxc/104/snapshot am 25.09.2026, dazu ein von Hand angelegter (älter, ohne Beschreibung).
|
|
PVESH_104 = [
|
|
{"description": UPDATE + "\n", "name": GITEA_SNAP, "snaptime": 1790328400},
|
|
{"description": "You are here!", "digest": "e355", "name": "current", "parent": GITEA_SNAP, "running": 1},
|
|
{"name": "vor-dem-umbau", "snaptime": 1789000000},
|
|
]
|
|
|
|
|
|
def test_snapshot_details_aus_pvesh():
|
|
a = _ausfuehrer_modul()
|
|
assert a.snapshot_details(PVESH_104) == [
|
|
{"name": "vor-dem-umbau", "zeit": 1789000000, "beschreibung": None, "eigen": False},
|
|
{"name": GITEA_SNAP, "zeit": 1790328400, "beschreibung": UPDATE, "eigen": True}]
|
|
assert a.snapshot_details([{"name": "current"}]) == [] and a.snapshot_details(None) == []
|
|
assert a.sicherung_details([{"volid": ALT, "ctime": 1790328597, "size": 360932725, "notes": "mc2-sicherung: …"}]) == [
|
|
{"volid": ALT, "zeit": 1790328597, "groesse": 360932725}]
|
|
|
|
|
|
def test_bericht_behaelt_die_namenslisten_und_bringt_details():
|
|
a = _ausfuehrer_modul()
|
|
antworten = {
|
|
"/nodes/pve/storage": [{"storage": "local", "type": "dir", "content": "backup,iso", "active": 1, "shared": 0,
|
|
"avail": 60 * 1024 ** 3}],
|
|
"/nodes/pve/storage/local/content": [
|
|
{"volid": ALT, "vmid": 105, "ctime": 1790328597, "size": 360932725,
|
|
"notes": "mc2-sicherung: vor einem Update durch den Homelab Orchestrator (proxmox-backup-server)"},
|
|
{"volid": "local:backup/vzdump-lxc-105-2026_09_01-03_00_00.tar.zst", "vmid": 105, "ctime": 1788000000,
|
|
"size": 1, "notes": "von Hand"}],
|
|
"/cluster/resources": [
|
|
{"vmid": 104, "type": "lxc", "node": "pve", "name": "gitea", "status": "stopped", "tags": "community-script"},
|
|
{"vmid": 105, "type": "lxc", "node": "pve", "name": "proxmox-backup-server", "status": "stopped",
|
|
"tags": "backup;community-script"}],
|
|
"/nodes/pve/lxc/104/snapshot": PVESH_104,
|
|
"/nodes/pve/lxc/105/snapshot": [{"name": "current"}],
|
|
"/nodes/pve/lxc/104/config": {"rootfs": "local-lvm:vm-104-disk-0,size=8G"},
|
|
"/nodes/pve/lxc/105/config": {"rootfs": "local-lvm:vm-105-disk-0,size=10G", "mp0": "/mnt/qnap,mp=/mnt/ds"},
|
|
}
|
|
|
|
def pvesh(pfad, *parameter):
|
|
if pfad not in antworten:
|
|
raise RuntimeError(f"pvesh {pfad}: gibt es im Test nicht")
|
|
return antworten[pfad]
|
|
|
|
a.NODE = "pve"
|
|
a._pvesh = pvesh
|
|
a._laufen = lambda befehl, zeitlimit=60, sanft=False: (0, "")
|
|
gitea, pbs = a.bericht()["gaeste"]
|
|
# Die Namensliste bleibt, wie pvesh sie liefert (andere Stellen hängen daran) — die Details kommen dazu.
|
|
assert gitea["snapshots"] == [GITEA_SNAP, "vor-dem-umbau"]
|
|
assert [d["name"] for d in gitea["snapshot_details"]] == ["vor-dem-umbau", GITEA_SNAP]
|
|
assert pbs["sicherungen"] == [ALT] and pbs["snapshots"] == [] and pbs["snapshot_details"] == []
|
|
assert pbs["sicherung_details"] == [{"volid": ALT, "zeit": 1790328597, "groesse": 360932725}]
|
|
|
|
|
|
def test_anlegen_mit_festem_anlass():
|
|
a = _ausfuehrer_modul()
|
|
a._pvesh = lambda pfad, *p: [{"vmid": 104, "type": "lxc", "tags": "community-script;git"}]
|
|
befehle: list[list[str]] = []
|
|
a._laufen = lambda befehl, zeitlimit=60, sanft=False: befehle.append(befehl) or (0, "")
|
|
antwort = a.ausfuehren({"aktion": "snapshot", "parameter": {"vmid": 104, "anlass": "knopf"}})
|
|
assert antwort["code"] == 0 and antwort["text"].startswith("snapshot=mc2-")
|
|
assert befehle[-1][:3] == ["pct", "snapshot", "104"] and befehle[-1][-2:] == ["--description", KNOPF]
|
|
assert a.ausfuehren({"aktion": "snapshot", "parameter": {"vmid": 104}})["code"] == 0 # wie bisher: vor Update
|
|
assert befehle[-1][-1] == UPDATE
|
|
# Der Server wählt nur aus der Liste; ein freier Text kommt nicht in die Beschreibung.
|
|
antwort = a.ausfuehren({"aktion": "snapshot", "parameter": {"vmid": 104, "anlass": "; rm -rf /"}})
|
|
assert antwort["code"] == 2 and "Anlass" in antwort["text"] and len(befehle) == 2
|
|
|
|
|
|
# --- Homelab-Teil: die Liste -----------------------------------------------------------------------------------------
|
|
|
|
def _bericht(**gitea) -> dict:
|
|
"""Der echte Bericht vom 24.09. mit Snapshots und Sicherungen, wie der Ausführer sie seit dem 25.09. schickt."""
|
|
b = json.loads(json.dumps(BERICHT))
|
|
g = {x["vmid"]: x for x in b["gaeste"]}
|
|
g[104].update(snapshots=[GITEA_SNAP, "vor-dem-umbau"], snapshot_details=[
|
|
{"name": "vor-dem-umbau", "zeit": 1789000000, "beschreibung": None, "eigen": False},
|
|
{"name": GITEA_SNAP, "zeit": 1790328400, "beschreibung": UPDATE, "eigen": True}], **gitea)
|
|
g[105].update(sicherungen=[ALT], sicherung_details=[{"volid": ALT, "zeit": 1790328597, "groesse": 360932725}],
|
|
sicherung_moeglich=True, sicherung_speicher="local", snapshot_details=[])
|
|
g[106].update(snapshots=["docker-vorher"], snapshot_details=[
|
|
{"name": "docker-vorher", "zeit": 1789500000, "beschreibung": "vor dem Umzug", "eigen": False}])
|
|
for x in (100, 101, 102, 103):
|
|
g[x]["snapshot_details"] = []
|
|
b["gaeste"].append({"vmid": 107, "art": "lxc", "name": "homelab-orchestrator", "status": "running",
|
|
"etiketten": ["mc2"], "erlaubt": False, "snapshots": ["test"]})
|
|
return b
|
|
|
|
|
|
@pytest.fixture
|
|
def daten(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path))
|
|
monkeypatch.setenv("MC_NOTIFY_LOG", str(tmp_path / "notify.log"))
|
|
for name in ("MC_AUSFUEHRER_TOKEN", "MC_WAECHTER_STORE", "MC_ARCANE_KEY"):
|
|
monkeypatch.delenv(name, raising=False)
|
|
einstellungen_mod.einstellungen.cache_clear()
|
|
monkeypatch.setattr(kanal, "_kontakt", {"zuletzt": None})
|
|
monkeypatch.setattr(inventar, "erreichbar", lambda url: True)
|
|
monkeypatch.setattr(inventar.karenz, "skript_geaendert", lambda kennung: 0.0) # kein Netz
|
|
yield tmp_path
|
|
einstellungen_mod.einstellungen.cache_clear()
|
|
|
|
|
|
def _geraete(stand: dict) -> dict[str, dict]:
|
|
return {g["id"]: g for g in stand["geraete"]}
|
|
|
|
|
|
def test_liste_je_geraet_mit_herkunft_und_knoepfen(daten):
|
|
kanal.bericht_speichern(_bericht())
|
|
stand = snapshots.liste()
|
|
g = _geraete(stand)
|
|
assert list(g) == ["ct-100", "ct-101", "ct-102", "ct-103", "ct-104", "ct-105", "vm-106"] # ohne ct-107
|
|
assert stand["details"] is True and stand["sperre"] is None and stand["ausfuehrer"]["verbunden"] is True
|
|
assert stand["summe"] == {"snapshots": 3, "eigene": 1, "sicherungen": 1, "sicherungen_groesse": 360932725}
|
|
|
|
gitea = g["ct-104"]
|
|
assert (gitea["name"], gitea["wo"], gitea["art"], gitea["laeuft"], gitea["grund"]) == (
|
|
"Gitea", "Container 104", "container", None, None)
|
|
hand, eigen = gitea["snapshots"] # ältester zuerst
|
|
assert (hand["name"], hand["herkunft"], hand["eigen"], hand["loeschen"], hand["zuruecksetzen"]) == (
|
|
"vor-dem-umbau", "proxmox", False, None, None)
|
|
assert (eigen["herkunft"], eigen["zeit"], eigen["beschreibung"]) == ("update", 1790328400, UPDATE)
|
|
assert eigen["loeschen"]["pfad"] == f"/api/homelab/snapshots/ct-104/{GITEA_SNAP}/loeschen"
|
|
assert eigen["loeschen"]["frage"] == (
|
|
f"Den Snapshot {GITEA_SNAP} von Gitea (vom 25.09.2026 11:26, Rückweg eines Updates) endgültig löschen? Am Gerät "
|
|
"selbst ändert sich nichts; nur der Weg zurück auf diesen Stand ist danach weg.")
|
|
zurueck = eigen["zuruecksetzen"]
|
|
assert zurueck["pfad"].endswith(f"/{GITEA_SNAP}/zuruecksetzen") and zurueck["label"] == "Zurücksetzen"
|
|
assert zurueck["frage"].startswith(f"Gitea (Container 104) auf den Snapshot vom 25.09.2026 11:26 ({GITEA_SNAP}) "
|
|
"zurücksetzen? Alles, was sich seitdem im Container geändert hat")
|
|
assert "ist danach verloren" in zurueck["frage"] and "gestoppt und wieder gestartet" in zurueck["frage"]
|
|
assert gitea["anlegen"]["pfad"] == "/api/homelab/snapshots/ct-104/anlegen"
|
|
assert "Der Container läuft dabei weiter." in gitea["anlegen"]["frage"]
|
|
assert "räumt kein Update weg" in gitea["anlegen"]["frage"]
|
|
|
|
pbs = g["ct-105"]
|
|
assert pbs["anlegen"] is None and pbs["grund"] == (
|
|
"Kein Snapshot möglich: Bind-Mount mp0 (/mnt/qnap-backup/pbs-datastore) — nur Backup möglich. Vor einem Update "
|
|
"legt der Orchestrator stattdessen eine Sicherung an.")
|
|
(sicherung,) = pbs["sicherungen"]
|
|
assert (sicherung["datei"], sicherung["speicher"], sicherung["groesse"]) == (
|
|
"vzdump-lxc-105-2026_09_25-11_29_57.tar.zst", "local", 360932725)
|
|
assert sicherung["loeschen"]["pfad"] == (
|
|
"/api/homelab/sicherungen/ct-105/vzdump-lxc-105-2026_09_25-11_29_57.tar.zst/loeschen")
|
|
assert "(vom 25.09.2026 11:29, 361 MB, auf „local“)" in sicherung["loeschen"]["frage"]
|
|
|
|
# Arcane ohne Etikett (Stand 24.09.): zu sehen, aber ohne Knöpfe.
|
|
arcane = g["vm-106"]
|
|
assert arcane["anlegen"] is None and arcane["grund"].startswith("Nicht freigegeben")
|
|
assert arcane["snapshots"][0]["herkunft"] == "proxmox" and arcane["snapshots"][0]["loeschen"] is None
|
|
|
|
|
|
def test_rueckweg_testen_und_aufraeumen_in_der_liste(daten):
|
|
kanal.bericht_speichern(_bericht())
|
|
stand = snapshots.liste()
|
|
g = _geraete(stand)
|
|
probe = g["ct-104"]["probe"]
|
|
assert (probe["pfad"], probe["label"]) == ("/api/homelab/ziele/ct-104/rueckweg-probe", "Rückweg testen")
|
|
assert "legt einen Snapshot an, wertet die Prüfung absichtlich als rot" in probe["frage"]
|
|
assert "eine Sicherung an" in g["ct-105"]["probe"]["frage"] # PBS: Sicherung statt Snapshot
|
|
assert g["vm-106"]["probe"] is None # nicht freigegeben
|
|
auf = stand["aufraeumen"]
|
|
assert (auf["pfad"], auf["label"]) == ("/api/homelab/snapshots/rueckwege-loeschen", "Alle 2 Update-Rückwege löschen")
|
|
assert auf["frage"].startswith("1 Snapshot und 1 Sicherung endgültig löschen (Gitea, ")
|
|
assert "per Knopf oder von Hand in Proxmox angelegt hast, bleiben" in auf["frage"]
|
|
assert stand["aufraeumen_laeuft"] is False
|
|
|
|
|
|
def test_liste_mit_altem_ausfuehrer_nur_aus_namen(daten):
|
|
"""Bis der neue Ausführer läuft, gibt es nur Namen: der Zeitpunkt aus dem Namen, keine Beschreibung."""
|
|
b = json.loads(json.dumps(BERICHT))
|
|
next(g for g in b["gaeste"] if g["vmid"] == 104)["snapshots"] = [GITEA_SNAP, "vor-dem-umbau"]
|
|
next(g for g in b["gaeste"] if g["vmid"] == 105)["sicherungen"] = [ALT]
|
|
kanal.bericht_speichern(b)
|
|
stand = snapshots.liste()
|
|
assert stand["details"] is False
|
|
gitea = _geraete(stand)["ct-104"]
|
|
assert [(s["name"], s["zeit"], s["beschreibung"]) for s in gitea["snapshots"]] == [
|
|
("vor-dem-umbau", None, None), (GITEA_SNAP, 1790328399.0, None)]
|
|
(sicherung,) = _geraete(stand)["ct-105"]["sicherungen"]
|
|
assert sicherung["zeit"] == 1790328597.0 and sicherung["groesse"] is None
|
|
|
|
|
|
def test_ohne_bericht_leer(daten):
|
|
stand = snapshots.liste()
|
|
assert stand["geraete"] == [] and stand["bericht"] is None and stand["ausfuehrer"]["verbunden"] is False
|
|
|
|
|
|
# --- Aktionen mit einem nachgespielten Ausführer ---------------------------------------------------------------------
|
|
|
|
class Ausfuehrer:
|
|
"""Beantwortet Aufträge wie der echte und führt einen Bericht, der sich mit ihnen ändert."""
|
|
|
|
def __init__(self, bericht: dict):
|
|
self.bericht = bericht
|
|
self.auftraege: list[tuple[str, dict]] = []
|
|
self.fehlschlag: dict[str, tuple[int, str]] = {}
|
|
self.stopp = threading.Event()
|
|
threading.Thread(target=self._laufen, daemon=True).start()
|
|
|
|
def gast(self, vmid: int) -> dict:
|
|
return next(g for g in self.bericht["gaeste"] if g["vmid"] == vmid)
|
|
|
|
def _laufen(self) -> None:
|
|
while not self.stopp.is_set():
|
|
a = kanal.naechster()
|
|
if not a:
|
|
time.sleep(0.02)
|
|
continue
|
|
aktion, p = a["aktion"], a["parameter"]
|
|
self.auftraege.append((aktion, p))
|
|
code, text = self.fehlschlag.get(aktion, (0, ""))
|
|
if code == 0 and aktion == "snapshot":
|
|
g = self.gast(p["vmid"])
|
|
g["snapshots"].append(NEU)
|
|
# Ein Ausführer, der den Anlass noch nicht kennt, schreibt „Vor einem Update …“ hinein.
|
|
g["snapshot_details"].append({"name": NEU, "zeit": 1790333112, "beschreibung": UPDATE, "eigen": True})
|
|
text = f"snapshot={NEU}\nfreeze guest filesystem"
|
|
elif code == 0 and aktion == "snapshot_loeschen":
|
|
g = self.gast(p["vmid"])
|
|
g["snapshots"].remove(p["snapshot"])
|
|
g["snapshot_details"] = [d for d in g["snapshot_details"] if d["name"] != p["snapshot"]]
|
|
elif code == 0 and aktion == "sicherung_loeschen":
|
|
g = self.gast(p["vmid"])
|
|
g["sicherungen"].remove(p["sicherung"])
|
|
g["sicherung_details"] = [d for d in g["sicherung_details"] if d["volid"] != p["sicherung"]]
|
|
elif code == 0 and aktion == "bericht":
|
|
text = json.dumps(self.bericht)
|
|
kanal.ergebnis(a["id"], code, text)
|
|
|
|
|
|
_echtes_warten = kanal.warten
|
|
|
|
|
|
@pytest.fixture
|
|
def host(daten, monkeypatch):
|
|
monkeypatch.setattr(updates, "WARTEN_NACH_UPDATE_S", 0)
|
|
monkeypatch.setattr(kanal, "warten", lambda aid, zeitlimit_s, takt_s=2.0: _echtes_warten(aid, min(zeitlimit_s, 10),
|
|
0.02))
|
|
meldungen: list[tuple[str, str]] = []
|
|
monkeypatch.setattr(snapshots.announce, "notify_telegram", lambda betreff, text: meldungen.append((betreff, text)))
|
|
verlauf: list[tuple] = []
|
|
monkeypatch.setattr(updates.update_verlauf, "eintragen", lambda *a: verlauf.append(a))
|
|
bericht = _bericht()
|
|
kanal.bericht_speichern(bericht)
|
|
a = Ausfuehrer(json.loads(json.dumps(bericht)))
|
|
a.meldungen, a.verlauf = meldungen, verlauf
|
|
yield a
|
|
a.stopp.set()
|
|
|
|
|
|
def _auf_ende(antwort: dict) -> dict:
|
|
assert antwort["ok"], antwort
|
|
ende = time.time() + 20
|
|
while time.time() < ende:
|
|
lauf = updates.lauf_lesen(antwort["lauf"]) or {}
|
|
if lauf.get("status") == "fertig":
|
|
return lauf
|
|
time.sleep(0.05)
|
|
raise AssertionError("Lauf wurde nicht fertig")
|
|
|
|
|
|
def test_snapshot_anlegen_ohne_meldung_und_von_keinem_update_weggeraeumt(host):
|
|
lauf = _auf_ende(snapshots.anlegen("ct-104"))
|
|
assert (lauf["ergebnis"], lauf["baustein"], lauf["snapshot"]) == ("angelegt", "snapshot-anlegen", NEU)
|
|
assert lauf["text"] == f"Gitea: Snapshot {NEU} angelegt."
|
|
assert host.auftraege == [("snapshot", {"vmid": 104, "anlass": "knopf"}), ("bericht", {})]
|
|
assert host.meldungen == [] and host.verlauf == [] # kein Update, keine Meldung bei Erfolg
|
|
gitea = _geraete(snapshots.liste())["ct-104"]
|
|
neu = next(s for s in gitea["snapshots"] if s["name"] == NEU)
|
|
# Der Ausführer schrieb noch „Vor einem Update …“ — der Lauf weiß es besser.
|
|
assert neu["herkunft"] == "knopf" and neu["loeschen"]["frage"].count("auf Knopfdruck angelegt") == 1
|
|
assert gitea["zuletzt"] == {"ok": True, "ende": lauf["ende"], "text": f"Gitea: Snapshot {NEU} angelegt."}
|
|
assert snapshots.knopf_namen("ct-104", host.gast(104)) == {NEU}
|
|
|
|
|
|
def test_gruenes_update_laesst_snapshots_per_knopf_stehen(host):
|
|
"""Sonst hätte das nächste Update den Snapshot gelöscht, den der User eben selbst angelegt hat."""
|
|
g = host.gast(104)
|
|
g["snapshots"] = [GITEA_SNAP, "mc2-20260925-120000", "vor-dem-umbau"]
|
|
g["snapshot_details"] = [{"name": "mc2-20260925-120000", "zeit": 1790330400, "beschreibung": KNOPF, "eigen": True}]
|
|
g["app"]["version"] = "1.27.3" # der Bericht nach dem Update: neue Version
|
|
kanal.bericht_speichern(_bericht())
|
|
lauf = _auf_ende(updates.starten("ct-104", "app"))
|
|
assert lauf["ergebnis"] == "eingespielt"
|
|
geloescht = [p["snapshot"] for aktion, p in host.auftraege if aktion == "snapshot_loeschen"]
|
|
assert geloescht == [GITEA_SNAP] # der per Knopf und der von Hand bleiben
|
|
|
|
|
|
def test_snapshot_loeschen_nur_eigene(host):
|
|
antwort = snapshots.loeschen("ct-104", "vor-dem-umbau")
|
|
assert antwort["ok"] is False and "von Hand angelegte verwaltest du in Proxmox" in antwort["detail"]
|
|
assert "gibt es laut dem letzten Bericht nicht" in snapshots.loeschen("ct-104", "mc2-20200101-000000")["detail"]
|
|
lauf = _auf_ende(snapshots.loeschen("ct-104", GITEA_SNAP))
|
|
assert (lauf["ergebnis"], lauf["text"]) == ("geloescht", f"Gitea: Snapshot {GITEA_SNAP} gelöscht.")
|
|
assert host.auftraege == [("snapshot_loeschen", {"vmid": 104, "snapshot": GITEA_SNAP}), ("bericht", {})]
|
|
assert [s["name"] for s in _geraete(snapshots.liste())["ct-104"]["snapshots"]] == ["vor-dem-umbau"]
|
|
|
|
|
|
def test_zuruecksetzen_gruen_mit_meldung(host):
|
|
lauf = _auf_ende(snapshots.zuruecksetzen("ct-104", GITEA_SNAP))
|
|
assert lauf["ergebnis"] == "zurueckgesetzt" and lauf["zurueckgesetzt"] is True
|
|
assert [a for a, _ in host.auftraege] == ["zurueck", "bericht"]
|
|
assert host.auftraege[0][1] == {"vmid": 104, "snapshot": GITEA_SNAP}
|
|
text = f"Gitea: auf den Snapshot vom 25.09.2026 11:26 ({GITEA_SNAP}) zurückgesetzt, Prüfung grün."
|
|
assert host.meldungen == [(snapshots.BETREFF, text)]
|
|
assert host.verlauf == []
|
|
|
|
|
|
def test_zuruecksetzen_rot_meldet_dringend(host, monkeypatch):
|
|
monkeypatch.setattr(inventar, "erreichbar", lambda url: False)
|
|
lauf = _auf_ende(snapshots.zuruecksetzen("ct-104", GITEA_SNAP))
|
|
assert (lauf["ergebnis"], lauf["zurueckgesetzt"], lauf["dringend"]) == ("fehler", True, True)
|
|
betreff, text = host.meldungen[-1]
|
|
assert betreff == snapshots.BETREFF_ALARM and "zurückgesetzt, aber die Prüfung ist rot" in text
|
|
assert "antwortet nicht" in text
|
|
eintrag = next(e for e in protokoll.eintraege() if e["art"] == "snapshot")
|
|
assert (eintrag["titel"], eintrag["stufe"], eintrag["ziel"]) == ("Zurückgesetzt, Prüfung rot · Gitea", "fehler",
|
|
"ct-104")
|
|
|
|
|
|
def test_zuruecksetzen_scheitert_und_sagt_wie_es_dem_geraet_geht(host):
|
|
host.fehlschlag["zurueck"] = (1, "unable to rollback: snapshot is locked")
|
|
lauf = _auf_ende(snapshots.zuruecksetzen("ct-104", GITEA_SNAP))
|
|
assert lauf["ergebnis"] == "fehler" and not lauf.get("zurueckgesetzt")
|
|
betreff, text = host.meldungen[-1]
|
|
assert betreff == snapshots.BETREFF_ALARM
|
|
assert text.startswith(f"Gitea: Zurücksetzen auf den Snapshot vom 25.09.2026 11:26 ({GITEA_SNAP}) gescheitert")
|
|
assert "snapshot is locked" in text and text.endswith("Gitea läuft.")
|
|
|
|
|
|
def test_sicherung_loeschen(host):
|
|
assert "gibt es laut dem letzten Bericht nicht" in snapshots.sicherung_loeschen("ct-105", "x.tar.zst")["detail"]
|
|
lauf = _auf_ende(snapshots.sicherung_loeschen("ct-105", "vzdump-lxc-105-2026_09_25-11_29_57.tar.zst"))
|
|
assert lauf["ergebnis"] == "geloescht" and host.auftraege[0] == ("sicherung_loeschen", {"vmid": 105, "sicherung": ALT})
|
|
assert _geraete(snapshots.liste())["ct-105"]["sicherungen"] == []
|
|
|
|
|
|
def test_gescheitertes_loeschen_meldet_ohne_alarm(host):
|
|
host.fehlschlag["snapshot_loeschen"] = (2, "Abgelehnt: Gast 104 hat kein erlaubtes Etikett")
|
|
lauf = _auf_ende(snapshots.loeschen("ct-104", GITEA_SNAP))
|
|
assert lauf["ergebnis"] == "fehler" and host.meldungen[-1][0] == snapshots.BETREFF
|
|
assert "ließ sich nicht löschen" in host.meldungen[-1][1] and "kein erlaubtes Etikett" in host.meldungen[-1][1]
|
|
assert _geraete(snapshots.liste())["ct-104"]["zuletzt"]["ok"] is False
|
|
|
|
|
|
def test_sperren_wie_bei_den_updates(host, monkeypatch):
|
|
# Nicht freigegeben, der eigene Container, ein unbekanntes Gerät
|
|
assert snapshots.anlegen("vm-106")["detail"].startswith("Dieses Gerät ist nicht freigegeben")
|
|
assert snapshots.anlegen("ct-107")["detail"] == "Dieses Gerät steht nicht im Bericht des Ausführers."
|
|
assert snapshots.anlegen("ct-999")["ok"] is False
|
|
# Kein Snapshot möglich (PBS: Bind-Mount)
|
|
assert snapshots.anlegen("ct-105")["detail"].startswith("Kein Snapshot möglich: Bind-Mount mp0")
|
|
# Während „Alle aktualisieren“ — die Seite sagt denselben Satz über der Liste.
|
|
with monkeypatch.context() as m:
|
|
m.setattr(sammellauf, "laeuft", lambda: True)
|
|
assert snapshots.anlegen("ct-104")["detail"].startswith(updates.SAMMELLAUF_SPERRE)
|
|
assert snapshots.liste()["sperre"] == snapshots.anlegen("ct-104")["detail"]
|
|
# Läuft irgendwo ein Update, warten alle Aktionen (sonst stünden sie beim Ausführer hinter ihm an) …
|
|
updates._merken({"id": "u1", "ziel": "ct-100", "baustein": "os", "name": "AdGuard Home", "status": "laeuft",
|
|
"start": time.time(), "ende": None, "ergebnis": None, "text": None, "schritte": ["os_update …"]})
|
|
assert snapshots.zuruecksetzen("ct-104", GITEA_SNAP)["detail"].startswith(
|
|
"Gerade läuft ein Update (AdGuard Home).")
|
|
assert _geraete(snapshots.liste())["ct-100"]["laeuft"] == "Update läuft"
|
|
updates._merken({**updates.lauf_lesen("u1"), "status": "fertig", "ende": time.time(), "ergebnis": "eingespielt"})
|
|
# … eine Aktion dieser Seite dagegen sperrt nur ihr Gerät; dort lehnen beide Seiten ab.
|
|
updates._merken({"id": "x1", "ziel": "ct-104", "baustein": "snapshot-zurueck", "name": "Gitea", "status": "laeuft",
|
|
"start": time.time(), "ende": None, "ergebnis": None, "text": None, "schritte": ["zurueck: ok"]})
|
|
assert "läuft gerade schon etwas" in snapshots.loeschen("ct-104", GITEA_SNAP)["detail"]
|
|
assert updates.starten("ct-104", "os")["ok"] is False
|
|
assert snapshots.liste()["sperre"] is None
|
|
assert _geraete(snapshots.liste())["ct-104"]["laeuft"] == "Zurückgesetzt, wird geprüft"
|
|
assert "ct-104" in updates.laufende_ziele() # der Wächter lässt das Gerät in Ruhe
|
|
# Ausführer im Nur-Lesen-Modus
|
|
nur_lesen = _bericht()
|
|
nur_lesen["host"]["nur_lesen"] = True
|
|
kanal.bericht_speichern(nur_lesen)
|
|
assert "Nur-Lesen-Modus" in snapshots.anlegen("ct-102")["detail"]
|
|
assert host.auftraege == []
|
|
|
|
|
|
def test_protokoll_und_betreff(host):
|
|
_auf_ende(snapshots.anlegen("ct-104"))
|
|
_auf_ende(snapshots.zuruecksetzen("ct-104", GITEA_SNAP))
|
|
eintraege = [e for e in protokoll.eintraege() if e["art"] == "snapshot"]
|
|
assert [(e["titel"], e["stufe"]) for e in eintraege] == [("Zurückgesetzt, Prüfung grün · Gitea", "ok"),
|
|
("Snapshot angelegt · Gitea", "ok")]
|
|
assert eintraege[1]["details"].endswith("Ohne Meldung: Das Ergebnis steht auf der Seite Snapshots.")
|
|
# Die Auftrags-Einträge heißen nicht mehr „Alter …“: Gelöscht wird auch von Hand.
|
|
assert protokoll._AKTION["snapshot_loeschen"][0] == "Snapshot gelöscht"
|
|
bekannt = protokoll._betreffe(updates.laeufe(100), [])
|
|
assert bekannt[f"Gitea: auf den Snapshot vom 25.09.2026 11:26 ({GITEA_SNAP}) zurückgesetzt, Prüfung grün."] == (
|
|
snapshots.BETREFF)
|
|
|
|
|
|
def test_alle_rueckwege_loeschen_nacheinander_mit_einer_meldung(host, monkeypatch):
|
|
monkeypatch.setattr(snapshots, "ABFRAGE_TAKT_S", 0.02)
|
|
assert snapshots.rueckwege_loeschen() == {"ok": True, "anzahl": 2}
|
|
ende = time.time() + 20
|
|
while snapshots._AUFRAEUMEN.is_set() and time.time() < ende:
|
|
time.sleep(0.05)
|
|
geloescht = [(a, p.get("snapshot") or p.get("sicherung")) for a, p in host.auftraege
|
|
if a in ("snapshot_loeschen", "sicherung_loeschen")]
|
|
assert geloescht == [("snapshot_loeschen", GITEA_SNAP), ("sicherung_loeschen", ALT)]
|
|
assert host.meldungen == [(snapshots.BETREFF, "Update-Rückwege aufgeräumt: 2 von 2 gelöscht.")]
|
|
stand = snapshots.liste()
|
|
# Der von Hand angelegte Snapshot bleibt; danach gibt es nichts mehr aufzuräumen.
|
|
assert [s["name"] for s in _geraete(stand)["ct-104"]["snapshots"]] == ["vor-dem-umbau"]
|
|
assert stand["aufraeumen"] is None
|
|
assert snapshots.rueckwege_loeschen()["detail"] == "Es gibt keine Rückwege von Updates zu löschen."
|
|
|
|
|
|
def test_schnittstelle(host):
|
|
from routers import homelab
|
|
app = FastAPI()
|
|
app.include_router(homelab.router)
|
|
c = TestClient(app)
|
|
daten = c.get("/api/homelab/snapshots").json()
|
|
assert [g["id"] for g in daten["geraete"]][:2] == ["ct-100", "ct-101"] and daten["summe"]["eigene"] == 1
|
|
antwort = c.post("/api/homelab/snapshots/ct-104/vor-dem-umbau/loeschen")
|
|
assert antwort.status_code == 409 and "Proxmox" in antwort.json()["detail"]
|
|
assert c.post("/api/homelab/sicherungen/ct-105/fremd.tar.zst/loeschen").status_code == 409
|
|
# Aufräumen: ohne Rückwege 409 mit Grund, sonst startet es (hier gibt es zwei).
|
|
antwort = c.post("/api/homelab/snapshots/rueckwege-loeschen")
|
|
assert antwort.status_code == 200 and antwort.json() == {"ok": True, "anzahl": 2}
|
|
ende = time.time() + 20
|
|
while snapshots._AUFRAEUMEN.is_set() and time.time() < ende:
|
|
time.sleep(0.05)
|
|
antwort = c.post("/api/homelab/snapshots/ct-102/anlegen")
|
|
assert antwort.status_code == 200 and antwort.json()["ok"] is True
|
|
lauf = _auf_ende(antwort.json())
|
|
assert lauf["ergebnis"] == "angelegt" and lauf["name"] == "NetBird"
|