"""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"] == "Gitea: Snapshot mc2-20260924-192352" and "vom 24.09. 19:23" in tipps[2]["text"] 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"