homelab: eigener Reiter „Snapshots“ mit Verwaltung (anlegen, löschen, zurücksetzen)
Die Seite /homelab/snapshots (in der Seitenleiste nach „Updates“: Snapshots sind der Rückweg der Updates) zeigt je Gerät die Snapshots mit Zeitpunkt, Alter, Herkunft (vor Update / per Knopf / von Hand) und Beschreibung, beim PBS die Sicherungen des Orchestrators mit Größe. Knöpfe „Snapshot anlegen“, „Zurücksetzen“ und „Löschen“ gibt es nur für freigegebene Geräte und nur für mc2-Snapshots bzw. eigene Sicherungen; von Hand angelegte bleiben reine Ansicht. Die Rückfragen kommen vom Homelab-Teil (wie im Ziel-Modell); Zurücksetzen sagt vorher, was verloren geht und dass das Gerät neu startet. Am Handy (320–400 px) untereinander, ab md Knöpfe rechts, ab xl fünf Spalten. Entscheide: - Ausführer: kein neuer Befehl. Der Bericht bringt zusätzlich snapshot_details (snaptime, Beschreibung, eigen) und sicherung_details (ctime, Größe); snapshots und sicherungen bleiben unverändert. Keine Belegung je Snapshot — die kennt Proxmox bei LVM-Thin nicht, die Seite sagt das so. „snapshot“ nimmt einen Anlass (update oder knopf) und wählt damit eine feste Beschreibung, damit ein Snapshot auf Knopfdruck nicht „Vor einem Update“ heißt. Bis der neue Ausführer eingespielt ist, liest der Homelab-Teil den Zeitpunkt aus dem Namen. - Aktionen sind Läufe in homelab-laeufe.json (Mechanik aus updates.py): gleiche Sperren (START_SPERRE, je Gerät einer, SAMMELLAUF_SPERRE), Protokoll-Art „snapshot“ mit eigenen Titeln, der Wächter lässt ein Gerät im Zurücksetzen in Ruhe. Zusätzlich keine Aktion, solange irgendwo ein Update läuft: Der Ausführer arbeitet einen Auftrag nach dem anderen ab — hinter einem Update liefe die Aktion in ihr Zeitlimit und käme danach unbeobachtet doch noch dran. - Zurücksetzen prüft danach wie ein Update und meldet immer ([Homelab-Snapshot], bei Rot als Alarm); anlegen und löschen melden nur ihr Scheitern, das Ergebnis steht beim Gerät. Nicht im Update-Verlauf, nicht unter „Zuletzt eingespielt“. - Per Knopf angelegte Snapshots räumt kein grünes Update weg (updates._aufraeumen fragt snapshots.knopf_namen). - speicher.py: alle Snapshots in einem Vorschlag mit Verweis auf die Seite statt einer je Snapshot. - Nebenbei: Der Update-Verlauf stürzte bei einem Ergebnis ohne Eintrag ab (etwa „offen“ nach einem Docker-Probelauf); die Protokoll-Seite stürzte im Entwicklungsmodus ab (Konstante „Symbol“ gegen das Symbol.for des React Compilers); Auftragstitel „Alter Snapshot gelöscht“ heißt jetzt „Snapshot gelöscht“ (gelöscht wird auch von Hand). Tests: pytest test_homelab_snapshots.py (Ausführer-Details, Liste, Läufe mit nachgespieltem Ausführer, Sperren, Aufräumen, Schnittstelle), test_homelab_speicher.py; Vitest SnapshotListe, lib/snapshots, Speicher. Prüftor grün. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
2784c0ea32
commit
450c782ed3
@@ -0,0 +1,428 @@
|
||||
"""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_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_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
|
||||
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"
|
||||
Reference in New Issue
Block a user