"""Lucys Knöpfe für den Orchestrator (mcp/orchestrator_werkzeuge.py, seit 25.09.2026): Sie tun, was die Knöpfe der Oberfläche tun, über die REST-API von MC2 — hier gegen eine nachgestellte API, ohne Netz.""" import json import sys from pathlib import Path import httpx import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "mcp")) import orchestrator_werkzeuge as w ZIELE_BOX = {"ziele": [{"id": "ki-box", "name": "KI-Box", "erreichbar": True, "bausteine": [ {"id": "hermes", "name": "Hermes", "zustand": "neu", "kurz": "+120 Änderungen", "aktion": {"methode": "POST", "pfad": "/api/maintenance/hermes-update", "frage": "?", "label": "Aktualisieren"}}, {"id": "os", "name": "Betriebssystem", "zustand": "aktuell"}]}]} ZIELE_HL = {"ziele": [ {"id": "ct-102", "name": "NetBird", "erreichbar": True, "bausteine": [ {"id": "app", "name": "NetBird", "zustand": "neu", "verfuegbar": "0.79.0", "aktion": None, "grund": "Das Update-Skript wurde am 24.09. geändert; zur Sicherheit erst ab 26.09. 08:06."}, {"id": "os", "name": "Pakete", "zustand": "neu", "kurz": "48 Pakete", "aktion": {"methode": "POST", "pfad": "/api/homelab/ziele/ct-102/os-update", "frage": "?", "label": "Jetzt"}}]}, {"id": "pve", "name": "Proxmox-Host", "erreichbar": True, "bausteine": []}]} SPEICHER = {"pools": [{"speicher": "local-lvm", "anteil": 0.5, "stufe": "ok"}], "systemplatte": {"anteil": 0.32, "stufe": "ok"}, "nas": [{"server": "192.168.178.62", "anteil": 0.39, "stufe": "ok", "problem": None}], "sicherungen": {"stufe": "ok", "fehlend": []}, "empfehlungen": [{"titel": "Arcane (Docker): 125 GB an den Pool zurückgeben", "platz": 125e9, "text": "Discard einschalten."}]} class Antwort: def __init__(self, status: int, daten: object): self.status_code, self._daten = status, daten self.content = json.dumps(daten).encode() def json(self): return self._daten @pytest.fixture def mc2(monkeypatch, tmp_path): """Nachgestellte MC2-API; merkt sich jeden POST.""" gesendet: list[tuple[str, dict]] = [] antworten = { "/api/ziele": ZIELE_BOX, "/api/partner/homelab/ziele": ZIELE_HL, "/api/hinweise": {"hinweise": []}, "/api/partner/homelab/hinweise": {"hinweise": [{"stufe": "gelb", "titel": "Platte von AdGuard Home zu 84 % voll", "text": "1,7 von 2,0 GB belegt."}]}, "/api/partner/homelab/speicher": SPEICHER, "/api/jobs": {"jobs": [{"label": "Nach Updates suchen", "state": "running"}]}, "/api/partner/homelab/laeufe": {"laeufe": []}, "/api/partner/homelab/sammellauf": {"sammellauf": None}, } post_antwort = {"status": 200, "daten": {"ok": True, "lauf": "l1"}} def get(url, params=None, timeout=None): pfad = url.removeprefix(w.MC_URL) return Antwort(200, antworten[pfad]) if pfad in antworten else Antwort(404, {"detail": "gibt es nicht"}) def post(url, json=None, timeout=None): gesendet.append((url.removeprefix(w.MC_URL), json)) return Antwort(post_antwort["status"], post_antwort["daten"]) monkeypatch.setattr(w.httpx, "get", get) monkeypatch.setattr(w.httpx, "post", post) monkeypatch.setattr(w, "LOG", tmp_path / "orchestrator-werkzeuge.log") return gesendet, post_antwort, tmp_path / "orchestrator-werkzeuge.log" def test_lage_und_updates_lesen(mc2): lage = w.lage() assert "KI-Box: keine Hinweise." in lage and "Homelab, gelb: Platte von AdGuard Home zu 84 % voll." in lage assert "Geräte: 3 von 3 erreichbar; 3 Updates stehen an" in lage assert "Speicherpool local-lvm: 50 % belegt (ok)." in lage and "Läuft auf der KI-Box: Nach Updates suchen." in lage liste = w.updates() assert "NetBird (ziel=ct-102, baustein=os): Pakete · 48 Pakete — startbar." in liste assert "kein Knopf: Das Update-Skript wurde am 24.09. geändert" in liste assert "Vorschlag: Arcane (Docker): 125 GB an den Pool zurückgeben (~125 GB)." in w.speicher() def test_update_starten_wie_der_knopf(mc2): gesendet, _, log = mc2 text = w.update_starten("ct-102", "os", "Commander: spiel die NetBird-Pakete ein") assert text.startswith("Gestartet: NetBird · Pakete.") assert gesendet == [("/api/partner/homelab/ziele/ct-102/os-update", {})] # Durchreiche zum Homelab-Teil eintrag = json.loads(log.read_text(encoding="utf-8").splitlines()[-1]) assert eintrag["werkzeug"] == "update_starten" and eintrag["grund"].startswith("Commander:") assert "kurz nicht erreichbar" in w.update_starten("ki-box", "hermes", "ja, Hermes aktualisieren") assert gesendet[-1][0] == "/api/maintenance/hermes-update" def test_ohne_grund_oder_knopf_passiert_nichts(mc2): gesendet, antwort, _ = mc2 assert w.update_starten("ct-102", "os", " ").startswith("Nicht ausgeführt: Es fehlt der Grund") assert w.update_starten("ct-102", "app", "bitte").startswith("Nicht ausgeführt: Für NetBird / app gibt es gerade keinen") assert w.update_starten("ct-999", "os", "bitte").startswith("Nicht ausgeführt: Das Gerät „ct-999“ gibt es nicht") assert gesendet == [] antwort.update(status=409, daten={"detail": "Es läuft gerade ‚Alle aktualisieren‘."}) assert w.alle_aktualisieren("homelab", "alles bitte") == "Nicht ausgeführt: Es läuft gerade ‚Alle aktualisieren‘." antwort.update(status=200, daten={"ok": False, "detail": "Unbekannter Dienst"}) assert w.dienst_neustarten("rm -rf", "test") == "Nicht ausgeführt: Unbekannter Dienst" assert w.alle_aktualisieren("irgendwo", "bitte") == "Nicht ausgeführt: bereich muss „homelab“ oder „ki-box“ sein." def test_aktionen_tragen_die_leitplanke(): lesen = {"lage", "updates", "speicher", "protokoll", "radar"} for name, text in w.BESCHREIBUNG.items(): if name in lesen or name == "dienst_neustarten": # Reparatur darf sie selbst (wie der Wächter) assert "Nur aufrufen, wenn der Commander" not in text, name else: assert "Nur aufrufen, wenn der Commander" in text, name assert "Sag das dem Commander vorher und frag nach" in w.BESCHREIBUNG["host_neustarten"] def test_mc2_nicht_erreichbar(monkeypatch, tmp_path): def kaputt(*a, **k): raise httpx.ConnectError("weg") monkeypatch.setattr(w.httpx, "post", kaputt) monkeypatch.setattr(w, "LOG", tmp_path / "log") assert w.kernel_aufraeumen("bitte aufräumen") == "Nicht ausgeführt: MC2 ist nicht erreichbar (ConnectError)."