"""Plattengesundheit (Ausbauplan „Später“, Punkt 18, 26.09.2026): Proxmox-Host (pvesh), KI-Box (smartctl) und NAS (SNMP) in einer Form, gemeldet wird nur, was schlecht ist oder schlechter wird. Die Werte stammen von den echten Platten am 26.09. (PELADN 512GB im Host, YMTC PC41Q-2TB in der Box).""" import socket import threading import pytest from kern import einstellungen as einstellungen_mod from kern import platten as kp from services.homelab import nas_snmp NVME_TEXT = """ SMART/Health Information (NVMe Log 0x02, NSID 0xffffffff) Critical Warning: 0x00 Temperature: 41 Celsius Available Spare: 100% Available Spare Threshold: 1% Percentage Used: 2% Power On Hours: 9,992 Unsafe Shutdowns: 6 Media and Data Integrity Errors: 0 Error Information Log Entries: 0 """ HOST_DISK = {"devpath": "/dev/nvme0n1", "model": "PELADN 512GB", "serial": "2024110610000123", "type": "nvme", "size": 512110190592, "health": "PASSED", "wearout": 98} BOX_JSON = {"model_name": "YMTC PC41Q-2TB-B", "serial_number": "S1", "smart_status": {"passed": True}, "user_capacity": {"bytes": 2048408248320}, "temperature": {"current": 35}, "power_on_time": {"hours": 2380}, "nvme_smart_health_information_log": {"critical_warning": 0, "temperature": 35, "available_spare": 100, "available_spare_threshold": 10, "percentage_used": 0, "media_errors": 0, "power_on_hours": 2380}} @pytest.fixture def daten(tmp_path, monkeypatch): monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path)) einstellungen_mod.einstellungen.cache_clear() yield tmp_path einstellungen_mod.einstellungen.cache_clear() def test_host_platte_aus_proxmox(): p = kp.aus_proxmox(HOST_DISK, {"health": "PASSED", "type": "text", "text": NVME_TEXT}) assert (p["id"], p["name"], p["wo"], p["typ"]) == ("pve:2024110610000123", "PELADN 512GB", "Proxmox-Host /dev/nvme0n1", "nvme") assert (p["gesund"], p["verbraucht"], p["temperatur"], p["reserve"], p["reserve_grenze"], p["stunden"]) == ( True, 2, 41, 100, 1, 9992) assert p["fehler"] == {"medien": 0} and kp.bewerten(p, None) == ("ok", []) def test_sata_attribute(): werte = kp.aus_ata_attributen([{"id": 5, "raw": "8"}, {"id": 197, "raw": "0"}, {"id": 194, "raw": "36 (Min/Max 20/45)"}, {"id": "x", "raw": "1"}]) assert werte == {"fehler": {"umgelagert": 8, "ausstehend": 0}, "temperatur": 36} def test_box_platte_aus_smartctl(): p = kp.aus_smartctl(BOX_JSON, "/dev/nvme0n1", "KI-Box") assert (p["id"], p["wo"], p["typ"], p["verbraucht"], p["temperatur"], p["stunden"]) == ( "KI-Box:S1", "KI-Box /dev/nvme0n1", "nvme", 0, 35, 2380) @pytest.mark.parametrize("aenderung, stufe, satz", [ ({"gesund": False}, "rot", "meldet sich selbst als defekt (SMART)"), ({"warnung": 4}, "rot", "meldet eine kritische Warnung (0x04)"), ({"reserve": 9, "reserve_grenze": 10}, "rot", "Ersatzblöcke fast aufgebraucht (9 %)"), ({"verbraucht": 85}, "gelb", "zu 85 % verbraucht"), ({"verbraucht": 93}, "rot", "zu 93 % verbraucht — Ersatz einplanen"), ({"temperatur": 72}, "gelb", "72 °C warm (ab 70 °C zu heiß)"), ({"fehler": {"medien": 3}}, "gelb", "Medienfehler: 0 → 3"), ({"smart_warnung": True}, "gelb", "SMART meldet eine Warnung"), ]) def test_bewerten(aenderung, stufe, satz): p = {**kp.aus_smartctl(BOX_JSON, "/dev/nvme0n1", "KI-Box"), **aenderung} assert kp.bewerten(p, {"medien": 0}) == (stufe, [satz]) def test_nur_was_schlechter_wird(daten): alt = {**kp.aus_smartctl(BOX_JSON, "/dev/nvme0n1", "KI-Box"), "fehler": {"medien": 5}} assert kp.bewertet([alt])[0]["stufe"] == "ok" # beim ersten Blick schon da: Ausgangslage schlechter = {**alt, "fehler": {"medien": 7}} (p,) = kp.bewertet([schlechter]) assert (p["stufe"], p["saetze"]) == ("gelb", ["Medienfehler: 5 → 7"]) assert kp.befund_titel(p) == "Platte YMTC PC41Q-2TB-B (KI-Box /dev/nvme0n1): Medienfehler: 5 → 7" assert kp.hinnehmen(p["id"], [schlechter])["ok"] assert kp.bewertet([schlechter])[0]["stufe"] == "ok" assert not kp.hinnehmen("gibt-es-nicht", [schlechter])["ok"] # --- NAS über SNMP ---------------------------------------------------------------------------------------------- def _antwort_pdu(anfrage: bytes, zeilen: list[tuple[str, object]], community: str) -> bytes: """Eine Response-PDU mit den Werten ab der erfragten OID (wie ein NAS, das GetBulk beantwortet).""" _, nachricht, _ = nas_snmp._lies(anfrage, 0) _, _, pos = nas_snmp._lies(nachricht, 0) _, gemeinschaft, pos = nas_snmp._lies(nachricht, pos) if gemeinschaft.decode() != community: return b"" # falscher Zugangsname: SNMP schweigt _, pdu, _ = nas_snmp._lies(nachricht, pos) _, nr, p = nas_snmp._lies(pdu, 0) for _ in range(2): _, _, p = nas_snmp._lies(pdu, p) _, bindungen, _ = nas_snmp._lies(pdu, p) _, bindung, _ = nas_snmp._lies(bindungen, 0) _, oid, _ = nas_snmp._lies(bindung, 0) ab = nas_snmp._oid_text(oid) def schluessel(o: str) -> tuple: return tuple(int(t) for t in o.split(".")) folgende = sorted((z for z in zeilen if schluessel(z[0]) > schluessel(ab)), key=lambda z: schluessel(z[0]))[:25] teile = b"" for o, w in folgende: wert = nas_snmp._tlv(0x04, w.encode()) if isinstance(w, str) else nas_snmp._int(int(w)) teile += nas_snmp._tlv(0x30, nas_snmp._oid(o) + wert) if len(folgende) < 25: teile += nas_snmp._tlv(0x30, nas_snmp._oid(ab if not folgende else folgende[-1][0]) + b"\x82\x00") antwort = nas_snmp._tlv(0xA2, nas_snmp._tlv(0x02, nr) + nas_snmp._int(0) + nas_snmp._int(0) + nas_snmp._tlv(0x30, teile)) return nas_snmp._tlv(0x30, nas_snmp._int(1) + nas_snmp._tlv(0x04, gemeinschaft) + antwort) HD = nas_snmp.HD_TABELLE QNAP = [(f"{nas_snmp.SYS_DESCR}.0", "Linux TS-453D 5.10.60-qnap"), (f"{HD}.1.1", 1), (f"{HD}.1.2", 2), (f"{HD}.1.3", 3), (f"{HD}.2.1", "HDD1"), (f"{HD}.2.2", "HDD2"), (f"{HD}.2.3", "HDD3"), (f"{HD}.3.1", "34 C/93 F"), (f"{HD}.3.2", "36 C/96 F"), (f"{HD}.3.3", "--"), (f"{HD}.4.1", 0), (f"{HD}.4.2", 0), (f"{HD}.4.3", -5), (f"{HD}.5.1", "WDC WD40EFRX"), (f"{HD}.5.2", "ST4000VN008"), (f"{HD}.5.3", "--"), (f"{HD}.7.1", "GOOD"), (f"{HD}.7.2", "Abnormal"), (f"{HD}.7.3", "--")] @pytest.fixture def nas(monkeypatch): """Ein nachgestelltes NAS auf einem freien UDP-Port von 127.0.0.1.""" sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(("127.0.0.1", 0)) sock.settimeout(0.2) laeuft = threading.Event() laeuft.set() def dienen(): while laeuft.is_set(): try: anfrage, absender = sock.recvfrom(65535) except (TimeoutError, OSError): continue if antwort := _antwort_pdu(anfrage, QNAP, "geheim-lesen"): sock.sendto(antwort, absender) faden = threading.Thread(target=dienen, daemon=True) faden.start() monkeypatch.setattr(nas_snmp, "PORT", sock.getsockname()[1]) monkeypatch.setattr(nas_snmp, "ZEITLIMIT_S", 0.5) yield laeuft.clear() faden.join(timeout=2) sock.close() def test_nas_platten_ueber_snmp(nas): ergebnis = nas_snmp.lesen("127.0.0.1", "geheim-lesen") assert ergebnis["ok"] and ergebnis["system"] == "Linux TS-453D 5.10.60-qnap" platten = ergebnis["platten"] assert [(p["name"], p["wo"], p["gesund"], p["temperatur"]) for p in platten] == [ ("WDC WD40EFRX", "NAS HDD1", True, 34), ("ST4000VN008", "NAS HDD2", False, 36)] # leerer Schacht fällt weg assert kp.bewerten(platten[1], None)[0] == "rot" def test_nas_falscher_zugangsname_und_speichern(nas, daten, monkeypatch): monkeypatch.setattr(nas_snmp, "nas_adresse", lambda: "127.0.0.1") falsch = nas_snmp.setzen("public", "") assert not falsch["ok"] and "antwortet nicht über SNMP" in falsch["detail"] and "Nichts gespeichert" in falsch["detail"] assert nas_snmp.zugang() is None richtig = nas_snmp.setzen("geheim-lesen", "") assert richtig == {"ok": True, "text": "Das NAS antwortet (Linux TS-453D 5.10.60-qnap): 2 Platten, davon 1 gesund."} assert nas_snmp.zugang() == {"community": "geheim-lesen", "adresse": None} assert "geheim-lesen" not in str(nas_snmp.stand()) assert nas_snmp.zugang_loeschen() and nas_snmp.stand() == {"eingerichtet": False, "ok": False, "platten": [], "detail": None} def test_disk_tabelle_der_neueren_firmware(): d = nas_snmp.DISK_TABELLE werte = [(f"{d}.1.1", 1), (f"{d}.2.1", "Disk 1"), (f"{d}.5.1", "GOOD"), (f"{d}.6.1", 33), (f"{d}.8.1", "WDC WD40EFRX-68N32N0")] (p,) = nas_snmp.platten_aus_disk(werte) assert (p["name"], p["gesund"], p["temperatur"]) == ("WDC WD40EFRX-68N32N0", True, 33)