HitonabiandClaude Opus 5.5
deaf35e953
User 25.09.: „Lucy ist hier der komplette Systemadmin + Helfer“ – „ja, bau das mit Lucy und dem Orchestrator“. Lucy bekommt die Knöpfe der Oberfläche als MCP-Werkzeuge: lesen (lage, updates, speicher, protokoll, radar) und handeln (update_starten, alle_aktualisieren, nach_updates_suchen, dienst_neustarten, kernel_aufraeumen, host_neustarten, radar_testen, radar_suchen) – über dieselben MC2-Routen mit Snapshot, Prüfung, Rückweg und Sperren. Jede Aktion braucht einen Grund (die Bitte des Commanders) und wird protokolliert; die Beschreibungen verbieten Aktionen auf eigene Initiative (Reparatur-Neustart ausgenommen). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
122 lines
6.5 KiB
Python
122 lines
6.5 KiB
Python
"""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)."
|