Files
mission-control-v2/backend/tests/test_homelab_speicher.py
T
HitonabiandClaude Opus 5.5 450c782ed3 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>
2026-09-25 13:46:28 +02:00

290 lines
16 KiB
Python

"""Speicherpool, NAS und Sicherungen (25.09.2026). Die Zahlen stammen aus dem echten Bericht vom 25.09.:
local-lvm zu 49 % belegt, davon 161 GB die Platte der Arcane-VM, die innen nur 35 GB nutzt (Discard aus)."""
import importlib.util
import math
import subprocess
import time
from pathlib import Path
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from services.homelab import speicher
AUSFUEHRER = Path(__file__).resolve().parents[2] / "deploy" / "homelab" / "ausfuehrer.py"
JETZT = 1_790_330_000.0 # 25.09.2026 ~11:13
NACHT = 1_790_296_203 # 25.09.2026 02:30 — die Sicherungen der Nacht
def _ausfuehrer():
spec = importlib.util.spec_from_file_location("ausfuehrer_speicher", AUSFUEHRER)
modul = importlib.util.module_from_spec(spec)
spec.loader.exec_module(modul)
return modul
def _bericht(pool_anteil: float = 0.4916, meta: float = 0.0209, nas_anteil: float = 0.3945) -> dict:
gesamt = 374_538_764_288
return {
"host": {
"pools": [{"speicher": "local-lvm", "art": "lvmthin", "aktiv": True, "belegt": math.ceil(gesamt * pool_anteil),
"gesamt": gesamt, "meta_anteil": meta, "verwaist": []}],
"netzlaufwerke": [{"ziel": "/mnt/qnap-backup", "quelle": "192.168.178.62:/Backup", "art": "nfs",
"eingebunden": True, "server": "192.168.178.62", "zustand": "ok",
"platz": {"gesamt": 3_847_503_609_856, "belegt": int(3_847_503_609_856 * nas_anteil)}}],
"sicherung_jobs": [{"id": "j1", "zeitplan": "02:30", "speicher": "pbs-qnap", "an": True, "alle": True,
"vmids": [], "ausgenommen": []}],
"sicherungsziele": {"pbs-qnap": {"art": "pbs", "aktiv": True, "belegt": 1, "gesamt": 2,
"neueste": {"102": NACHT, "104": NACHT, "106": NACHT}}},
},
"gaeste": [
{"vmid": 102, "art": "lxc", "name": "netbird", "app": {"kennung": "netbird"}, "pool_belegt": 2_076_616_687,
"platte": {"belegt": 888_725_504, "gesamt": 4_143_677_440}, "snapshots": [], "ungenutzt": []},
{"vmid": 104, "art": "lxc", "name": "gitea", "app": {"kennung": "gitea"}, "pool_belegt": 6_442_450_944,
"platte": {"belegt": 4_867_121_152, "gesamt": 8_350_298_112}, "snapshots": ["mc2-20260924-192352"],
"ungenutzt": []},
{"vmid": 106, "art": "qemu", "name": "arcane", "pool_belegt": 160_948_530_708, "discard": False,
"platte": {"belegt": 35_454_582_784, "gesamt": 151_205_154_816}, "snapshots": [], "ungenutzt": []},
],
}
# --- Ausführer: reine Auswertungen --------------------------------------------------------------------------------
def test_thin_pool_mit_echter_belegung_je_volume():
a = _ausfuehrer()
status = [{"storage": "local-lvm", "type": "lvmthin", "active": 1, "used": 184_123_256_523, "total": 374_538_764_288},
{"storage": "local", "type": "dir", "active": 1, "used": 1, "total": 2}]
konfig = [{"storage": "local-lvm", "type": "lvmthin", "vgname": "pve", "thinpool": "data"}]
lvs = [{"vg_name": "pve", "lv_name": "data", "lv_size": "374538764288", "data_percent": "49.16",
"metadata_percent": "2.09", "pool_lv": "", "origin": ""},
{"vg_name": "pve", "lv_name": "vm-106-disk-0", "lv_size": "161061273600", "data_percent": "99.93",
"metadata_percent": "", "pool_lv": "data", "origin": ""},
{"vg_name": "pve", "lv_name": "snap_vm-104-disk-0_mc2-20260924-192352", "lv_size": "8589934592",
"data_percent": "", "metadata_percent": "", "pool_lv": "data", "origin": "vm-104-disk-0"},
{"vg_name": "pve", "lv_name": "root", "lv_size": "103079215104", "data_percent": "", "pool_lv": "", "origin": ""}]
pools, zuteilung = a.thin_pools(status, konfig, lvs)
assert pools == [{"speicher": "local-lvm", "art": "lvmthin", "aktiv": True, "belegt": 184_123_256_523,
"gesamt": 374_538_764_288, "meta_anteil": pytest.approx(0.0209), "verwaist": []}]
# Der Snapshot teilt seine Blöcke mit dem Original und zählt nicht mit.
assert zuteilung == {"local-lvm:vm-106-disk-0": int(161_061_273_600 * 0.9993)}
def test_volumes_discard_und_ungenutzte_platten():
a = _ausfuehrer()
assert a.gast_volumes({"rootfs": "local-lvm:vm-105-disk-0,size=10G",
"mp0": "/mnt/qnap-backup/pbs-datastore,mp=/mnt/datastore"}) == (
["local-lvm:vm-105-disk-0"], [], True)
vm = {"scsi0": "local-lvm:vm-106-disk-0,iothread=1,size=150G", "ide2": "none,media=cdrom",
"efidisk0": "local-lvm:vm-106-disk-1,efitype=4m,size=4M", "unused0": "local-lvm:vm-106-disk-2", "cores": 4}
assert a.gast_volumes(vm) == (["local-lvm:vm-106-disk-0", "local-lvm:vm-106-disk-1"], ["local-lvm:vm-106-disk-2"],
False)
vm["scsi0"] += ",discard=on"
assert a.gast_volumes(vm)[2] is True
def test_vm_belegung_ueber_den_gast_agenten():
a = _ausfuehrer()
daten = {"result": [
{"name": "sda2", "type": "ext4", "mountpoint": "/", "used-bytes": 35_000_000_000, "total-bytes": 150_000_000_000},
{"name": "sda2", "type": "ext4", "mountpoint": "/var/snap", "used-bytes": 35_000_000_000,
"total-bytes": 150_000_000_000},
{"name": "sda1", "type": "vfat", "mountpoint": "/boot/efi", "used-bytes": 6_000_000, "total-bytes": 1_000_000_000},
{"name": "overlay", "type": "overlay", "mountpoint": "/var/lib/docker/x", "used-bytes": 9, "total-bytes": 9},
{"name": "loop3", "type": "squashfs", "mountpoint": "/snap/core", "used-bytes": 9, "total-bytes": 9}]}
assert a.fs_belegung(daten) == {"belegt": 35_006_000_000, "gesamt": 151_000_000_000}
assert a.fs_belegung({"result": []}) is None and a.fs_belegung(None) is None
def test_netzlaufwerke_eingebunden_und_fehlend():
a = _ausfuehrer()
findmnt = {"filesystems": [{"target": "/mnt/qnap-backup", "source": "192.168.178.62:/Backup", "fstype": "nfs"}]}
fstab = ("192.168.178.62:/Backup /mnt/qnap-backup nfs vers=3,hard,_netdev 0 0\n"
"//192.168.178.62/Medien /mnt/medien cifs credentials=/root/.smb,_netdev 0 0\n"
"# 192.168.178.9:/alt /mnt/alt nfs defaults 0 0\n"
"192.168.178.62:/Selten /mnt/selten nfs noauto 0 0\n"
"/dev/pve/root / ext4 errors=remount-ro 0 1\n")
assert a.netzlaufwerke(findmnt, fstab) == [
{"ziel": "/mnt/medien", "quelle": "//192.168.178.62/Medien", "art": "cifs", "eingebunden": False,
"server": "192.168.178.62"},
{"ziel": "/mnt/qnap-backup", "quelle": "192.168.178.62:/Backup", "art": "nfs", "eingebunden": True,
"server": "192.168.178.62"}]
def test_sicherungsauftraege_und_juengste_sicherung():
a = _ausfuehrer()
assert a.sicherungs_jobs([{"id": "j1", "schedule": "02:30", "storage": "pbs-qnap", "all": 1, "type": "vzdump"},
{"id": "j2", "schedule": "sat 03:00", "storage": "local", "vmid": "100,101",
"exclude": "", "enabled": 0}]) == [
{"id": "j1", "zeitplan": "02:30", "speicher": "pbs-qnap", "an": True, "alle": True, "vmids": [], "ausgenommen": []},
{"id": "j2", "zeitplan": "sat 03:00", "speicher": "local", "an": False, "alle": False, "vmids": [100, 101],
"ausgenommen": []}]
assert a.neueste_sicherungen([{"vmid": 100, "ctime": 5}, {"vmid": 100, "ctime": 9}, {"vmid": "101", "ctime": 7},
{"volid": "kaputt"}]) == {"100": 9, "101": 7}
def test_haengender_nfs_mount_haelt_den_bericht_nicht_auf(monkeypatch):
a = _ausfuehrer()
monkeypatch.setattr(a, "NETZ_ZEITLIMIT_S", 0.05)
gestartet: list[list[str]] = []
class Haengt:
returncode = None
def __init__(self, befehl, **_):
gestartet.append(befehl)
def poll(self):
return None
def kill(self):
pass # im Zustand D wirkt das nicht — der Prozess bleibt
monkeypatch.setattr(subprocess, "Popen", Haengt)
anfang = time.monotonic()
assert a._netz_platz("/mnt/qnap-backup") == ("haengt", None)
assert a._netz_platz("/mnt/qnap-backup") == ("haengt", None)
assert len(gestartet) == 1 and time.monotonic() - anfang < 2 # kein zweiter Prozess, kein Warten
# --- Homelab-Teil: Auswertung, Empfehlungen, Hinweise ---------------------------------------------------------
def test_pool_stufen_fuer_daten_und_metadaten():
assert speicher.pools(_bericht(0.79))[0]["stufe"] == "ok"
assert speicher.pools(_bericht(0.80))[0]["stufe"] == "gelb"
assert speicher.pools(_bericht(0.90))[0]["stufe"] == "rot" # 10 % frei (User-Vorgabe)
assert speicher.pools(_bericht(0.40, meta=0.91))[0]["stufe"] == "rot"
p = speicher.pools(_bericht())[0]
assert (p["speicher"], p["frei"]) == ("local-lvm", p["gesamt"] - p["belegt"])
def test_empfehlungen_groesster_gewinn_zuerst():
tipps = speicher.empfehlungen(_bericht())
assert [t["art"] for t in tipps] == ["trim-vm", "trim", "snapshot"]
arcane = tipps[0]
assert arcane["platz"] == 160_948_530_708 - 35_454_582_784 and arcane["titel"].startswith("Arcane (Docker): 125 GB")
assert "„Discard“ anhaken" in arcane["text"] and arcane["kurz"].endswith("(Discard einschalten)")
container = tipps[1]
assert container["platz"] == (2_076_616_687 - 888_725_504) + (6_442_450_944 - 4_867_121_152)
assert "pct fstrim 104 --ignore-mountpoints" in container["text"] and "Gitea 1,6 GB" in container["text"]
assert tipps[2]["titel"] == "1 Snapshot des Orchestrators — verwalten unter Homelab → Snapshots"
assert tipps[2]["text"].startswith("Bei Gitea.") and tipps[2]["platz"] is None
def test_snapshots_sind_ein_vorschlag():
"""25.09.2026: Nach „Alle aktualisieren“ standen fünf Snapshot-Vorschläge untereinander — jetzt einer, der auf die
Seite Snapshots zeigt. Von Hand angelegte zählen mit, der Container des Orchestrators selbst nicht."""
b = _bericht()
b["gaeste"][0]["snapshots"] = ["mc2-20260925-112118", "vor-dem-umbau"]
b["gaeste"][1]["snapshots"] = ["mc2-20260924-192352", "mc2-20260925-112639"]
b["gaeste"].append({"vmid": 107, "art": "lxc", "name": "homelab-orchestrator", "etiketten": ["mc2"],
"snapshots": ["test"], "ungenutzt": []})
tipps = [t for t in speicher.empfehlungen(b) if t["art"] == "snapshot"]
assert len(tipps) == 1
s = tipps[0]
assert s["titel"] == "3 Snapshots des Orchestrators — verwalten unter Homelab → Snapshots"
assert s["text"].startswith("Bei NetBird und Gitea (2).")
assert "Dazu 1 von Hand in Proxmox angelegt (NetBird)." in s["text"] and "homelab" not in s["text"]
assert s["kurz"] == "4 Snapshots (Homelab → Snapshots)"
b["gaeste"][1]["snapshots"] = []
b["gaeste"][0]["snapshots"] = ["vor-dem-umbau"]
assert [t["titel"] for t in speicher.empfehlungen(b) if t["art"] == "snapshot"] == [
"1 von Hand angelegter Snapshot — ansehen unter Homelab → Snapshots"]
b["gaeste"][0]["snapshots"] = []
assert all(t["art"] != "snapshot" for t in speicher.empfehlungen(b))
def test_kleine_gewinne_sind_keine_empfehlung():
b = _bericht()
b["gaeste"] = [b["gaeste"][0]] # NetBird allein: 1,2 GB — über der Schwelle, weil ≥ 1 GB
assert [t["art"] for t in speicher.empfehlungen(b)] == ["trim"]
b["gaeste"][0]["pool_belegt"] = b["gaeste"][0]["platte"]["belegt"] + 600_000_000
assert speicher.empfehlungen(b) == []
def test_nas_antwortet_nicht_haengt_oder_ist_voll():
assert speicher.nas(_bericht(), pruefen=lambda s, p: True)[0]["stufe"] == "ok"
weg = speicher.nas(_bericht(), pruefen=lambda s, p: False)[0]
assert (weg["stufe"], weg["problem"]) == ("rot", "Das NAS (192.168.178.62) antwortet nicht.")
b = _bericht()
b["host"]["netzlaufwerke"][0].update(zustand="haengt", platz=None)
assert "hängt" in speicher.nas(b, pruefen=lambda s, p: True)[0]["problem"]
assert speicher.nas(_bericht(nas_anteil=0.85), pruefen=lambda s, p: True)[0]["stufe"] == "gelb"
geprueft = []
speicher.nas(_bericht(), pruefen=lambda s, p: geprueft.append((s, p)) or True)
assert geprueft == [("192.168.178.62", 2049)]
def test_sicherungen_frisch_fehlend_und_ziel_weg():
s = speicher.sicherungen(_bericht(), JETZT)
assert (s["stufe"], s["erfasst"], s["fehlend"], s["zuletzt"]) == ("ok", 3, [], NACHT)
b = _bericht()
b["host"]["sicherungsziele"]["pbs-qnap"]["neueste"]["104"] = NACHT - 3 * 86400
s = speicher.sicherungen(b, JETZT)
assert s["stufe"] == "gelb" and [f["name"] for f in s["fehlend"]] == ["Gitea"]
# Wöchentlicher Auftrag: drei Tage alt ist in Ordnung.
b["host"]["sicherung_jobs"][0]["zeitplan"] = "sat 02:30"
assert speicher.sicherungen(b, JETZT)["stufe"] == "ok"
b = _bericht()
b["host"]["sicherungsziele"]["pbs-qnap"]["neueste"] = {}
assert speicher.sicherungen(b, JETZT)["stufe"] == "rot" # keiner gesichert: Auftrag läuft nicht
b = _bericht()
b["host"]["sicherungsziele"]["pbs-qnap"]["aktiv"] = False
s = speicher.sicherungen(b, JETZT)
assert s["stufe"] == "rot" and s["probleme"] == ["Das Sicherungsziel pbs-qnap ist nicht erreichbar."]
b["host"]["sicherung_jobs"][0]["ausgenommen"] = [102, 104, 106]
assert speicher.sicherungen(b, JETZT)["erfasst"] == 0
b["host"]["sicherung_jobs"] = []
assert speicher.sicherungen(b, JETZT)["stufe"] == "keine"
def test_waechter_befunde(monkeypatch):
monkeypatch.setattr(speicher, "antwortet", lambda s, p: True)
monkeypatch.setattr(speicher, "_bericht", lambda: _bericht())
assert speicher.pruefe_speicher(JETZT) == []
monkeypatch.setattr(speicher, "_bericht", lambda: _bericht(0.91, nas_anteil=0.93))
befunde = {b.id: b for b in speicher.pruefe_speicher(JETZT)}
pool = befunde["pool:local-lvm"]
assert (pool.stufe, pool.titel) == ("rot", "Speicherpool local-lvm zu 91 % voll")
assert "Platz bringt: 125 GB bei Arcane (Docker) per TRIM (Discard einschalten)" in pool.text
assert befunde["nas:/mnt/qnap-backup"].titel == "NAS (192.168.178.62) zu 93 % voll"
monkeypatch.setattr(speicher, "_bericht", dict)
assert speicher.pruefe_speicher(JETZT) == []
def test_schnittstelle(monkeypatch):
from routers import homelab
monkeypatch.setattr(speicher, "antwortet", lambda s, p: True)
monkeypatch.setattr(speicher, "_bericht", lambda: _bericht())
app = FastAPI()
app.include_router(homelab.router)
daten = TestClient(app).get("/api/homelab/speicher").json()
assert daten["bericht"] is True and daten["pools"][0]["stufe"] == "ok"
assert daten["nas"][0]["erreichbar"] is True and daten["sicherungen"]["stufe"] == "ok"
assert daten["empfehlungen"][0]["art"] == "trim-vm"
def test_alte_kernel_mit_knopf_und_systemplatte(monkeypatch):
b = _bericht()
b["host"].update(
kernel_alt=[{"paket": f"proxmox-kernel-6.17.{i}-1-pve-signed", "version": f"6.17.{i}-1-pve", "groesse": 915_000_000}
for i in range(20)],
kernel_bleiben=["6.14.8-2-pve", "6.17.13-21-pve", "7.0.14-12-pve", "7.0.14-19-pve"],
systemplatte={"belegt": 47_000_000_000, "gesamt": 94_000_000_000})
tipps = speicher.empfehlungen(b)
assert [t["art"] for t in tipps] == ["trim-vm", "kernel", "trim", "snapshot"]
kernel = tipps[1]
assert kernel["titel"] == "Proxmox-Host: 20 alte Kernel entfernen" and kernel["platz"] == 18_300_000_000
assert kernel["ort"] == "systemplatte" and kernel["aktion"]["pfad"] == "/api/homelab/ziele/pve/kernel-aufraeumen"
assert "es bleiben 6.14.8-2, 6.17.13-21, 7.0.14-12, 7.0.14-19" in kernel["text"]
assert speicher.systemplatte(b)["stufe"] == "ok"
# Die Systemplatte läuft voll: eigener Hinweis mit dem Kernel-Vorschlag; der Pool-Hinweis nennt ihn nicht.
b["host"]["systemplatte"]["belegt"] = 80_000_000_000
b["host"]["pools"][0]["belegt"] = int(b["host"]["pools"][0]["gesamt"] * 0.95)
monkeypatch.setattr(speicher, "antwortet", lambda s, p: True)
monkeypatch.setattr(speicher, "_bericht", lambda: b)
befunde = {x.id: x for x in speicher.pruefe_speicher(JETZT)}
assert befunde["host-platte"].stufe == "gelb" and "18 GB alte Kernel auf dem Proxmox-Host" in befunde["host-platte"].text
assert "Kernel" not in befunde["pool:local-lvm"].text