"""Einstellungen des Homelab-Teils (services/homelab/einstellungen.py, 24.09.2026): den Arcane-Schlüssel über die Oberfläche eintragen (erst von Arcane geprüft, nie zurückgegeben) und echte Docker-Updates statt Probelauf. Arcane ist ein kleiner Nachbau mit derselben Antwortform wie v2.12/2.13 (401 bei falschem Schlüssel).""" import json import logging import os import subprocess import sys import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from kern import einstellungen as einstellungen_mod from services.homelab import arcane, einstellungen, inventar, kanal BERICHT = json.loads((Path(__file__).parent / "fixtures" / "pve-bericht.json").read_text(encoding="utf-8")) BACKEND = Path(__file__).resolve().parents[1] RICHTIG = "arc_RichtigerSchluessel_0123456789abcdef" FALSCH = "arc_FalscherSchluessel_9876543210fedcba" ARCANE_URL_AUS_DEM_BERICHT = einstellungen.arcane_url # das Original; die Tests lenken auf den Nachbau um @pytest.fixture def arcane_nachbau(): """Arcane mit einer Umgebung und zwei Images, eins davon mit neuerem Stand.""" stand: dict = {"schluessel": RICHTIG, "status": None, "aufrufe": []} class Handler(BaseHTTPRequestHandler): def _antwort(self, code: int, daten: dict) -> None: roh = json.dumps(daten).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(roh))) self.end_headers() self.wfile.write(roh) def do_GET(self): stand["aufrufe"].append(self.path) if stand["status"]: return self._antwort(stand["status"], {"title": "Fehler", "status": stand["status"]}) if self.headers.get("X-API-Key") != stand["schluessel"]: return self._antwort(401, {"title": "Unauthorized", "status": 401, "detail": "Unauthorized: invalid API key"}) if self.path.startswith("/api/environments?"): return self._antwort(200, {"success": True, "data": [{"id": "0", "name": "local"}], "pagination": {"totalItems": 1}}) if self.path.startswith("/api/environments/0/images"): return self._antwort(200, {"success": True, "pagination": {"totalItems": 2}, "data": [ {"id": "a", "repo": "ghcr.io/tobi/nerdquiz", "tag": "latest", "usedBy": [{"name": "nerdquiz"}], "updateInfo": {"hasUpdate": True, "currentVersion": "1.0", "latestVersion": "1.1"}}, {"id": "b", "repo": "nginx", "tag": "1.27", "usedBy": [], "updateInfo": {"hasUpdate": False}}]}) return self._antwort(404, {"title": "Not Found", "status": 404}) def log_message(self, *args): pass server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) threading.Thread(target=server.serve_forever, daemon=True).start() yield f"http://127.0.0.1:{server.server_port}", stand server.shutdown() @pytest.fixture def daten(tmp_path, monkeypatch, arcane_nachbau): monkeypatch.setenv("MC_DATEN_DIR", str(tmp_path)) for name in ("MC_AUSFUEHRER_TOKEN", "MC_ARCANE_KEY", "MC_ARCANE_ECHT", "MC_HOMELAB_KARENZ_H"): monkeypatch.delenv(name, raising=False) einstellungen_mod.einstellungen.cache_clear() monkeypatch.setattr(kanal, "_kontakt", {"zuletzt": None}) arcane.cache_leeren() kanal.bericht_speichern(BERICHT) url, _ = arcane_nachbau monkeypatch.setattr(einstellungen, "arcane_url", lambda: url) # die VM 106 zeigt hier auf den Nachbau yield tmp_path arcane.cache_leeren() einstellungen_mod.einstellungen.cache_clear() def _client() -> TestClient: from routers import homelab as homelab_router app = FastAPI() app.include_router(homelab_router.router) return TestClient(app) def test_stand_von_anfang_an(daten, monkeypatch): monkeypatch.setattr(einstellungen, "arcane_url", lambda: None) assert einstellungen.stand()["arcane"]["url"] == "" monkeypatch.setattr(einstellungen, "arcane_url", ARCANE_URL_AUS_DEM_BERICHT) # die VM 106 aus dem Bericht stand = _client().get("/api/homelab/einstellungen").json() assert stand == { "arcane": {"url": "http://192.168.178.28:3552", "schluessel": "fehlt", "echt": False, "echt_quelle": "standard"}, "karenz_h": 48, "ausfuehrer": {"verbunden": True, "zuletzt": stand["ausfuehrer"]["zuletzt"], "nur_lesen": False, "sicherung_speicher": None}} bericht = json.loads(json.dumps(BERICHT)) bericht["host"].update(nur_lesen=True, sicherung={"speicher": "local", "art": "dir", "frei": 1}) kanal.bericht_speichern(bericht) monkeypatch.setenv("MC_HOMELAB_KARENZ_H", "72") stand = einstellungen.stand() assert stand["ausfuehrer"]["nur_lesen"] is True and stand["ausfuehrer"]["sicherung_speicher"] == "local" assert stand["karenz_h"] == 72 def test_richtiger_schluessel_wird_gespeichert_und_gilt_sofort(daten, arcane_nachbau, caplog): url, stand = arcane_nachbau caplog.set_level(logging.DEBUG) antwort = _client().post("/api/homelab/einstellungen/arcane-schluessel", json={"schluessel": f" {RICHTIG}\n"}) assert antwort.status_code == 200 assert antwort.json() == {"ok": True, "text": "Arcane nimmt den Schlüssel an: 2 Images in 1 Umgebung geprüft, 1 " "davon mit neuerem Stand. Die Übersicht zeigt die Docker-Images ab " "sofort."} assert stand["aufrufe"] == ["/api/environments?limit=50", "/api/environments/0/images?limit=500"] datei = daten / "arcane.key" assert datei.read_text(encoding="utf-8") == RICHTIG + "\n" and arcane.schluessel() == RICHTIG if os.name == "posix": assert datei.stat().st_mode & 0o077 == 0 assert _client().get("/api/homelab/einstellungen").json()["arcane"]["schluessel"] == "hinterlegt" # Ohne Neustart: Die Übersicht fragt Arcane mit dem neuen Schlüssel und sieht die Images. _, docker = inventar._arcane_bausteine(url, "vm-106", {}) assert (docker.zustand, docker.kurz, docker.aktion.label) == ("neu", "1 Images", "Probelauf") assert RICHTIG not in caplog.text def test_falscher_schluessel_wird_nicht_gespeichert(daten, caplog): caplog.set_level(logging.DEBUG) antwort = _client().post("/api/homelab/einstellungen/arcane-schluessel", json={"schluessel": FALSCH}) assert antwort.status_code == 200 assert antwort.json() == {"ok": False, "detail": "Arcane lehnt den Schlüssel ab (HTTP 401)."} assert not (daten / "arcane.key").exists() and arcane.schluessel() == "" assert FALSCH not in antwort.text and FALSCH not in caplog.text def test_ohne_rechte_und_bei_anderen_fehlern_nichts_gespeichert(daten, arcane_nachbau): _, stand = arcane_nachbau stand["status"] = 403 assert einstellungen.arcane_schluessel_setzen(RICHTIG)["detail"] == ( "Arcane lehnt den Schlüssel ab (HTTP 403). Ihm fehlt das Recht, Umgebungen und Images zu lesen.") stand["status"] = 500 assert einstellungen.arcane_schluessel_setzen(RICHTIG)["detail"] == ( "Arcane antwortet mit HTTP 500. Der Schlüssel ist nicht gespeichert.") assert not (daten / "arcane.key").exists() def test_arcane_nicht_erreichbar(daten, monkeypatch): frei = ThreadingHTTPServer(("127.0.0.1", 0), BaseHTTPRequestHandler) tot = f"http://127.0.0.1:{frei.server_port}" frei.server_close() # der Port ist jetzt zu monkeypatch.setattr(einstellungen, "arcane_url", lambda: tot) ergebnis = einstellungen.arcane_schluessel_setzen(RICHTIG) assert ergebnis == {"ok": False, "detail": f"Arcane ist nicht erreichbar ({tot}). Der Schlüssel ist nicht " "gespeichert."} monkeypatch.setattr(einstellungen, "arcane_url", lambda: None) assert einstellungen.arcane_schluessel_setzen(RICHTIG)["detail"].startswith("Arcane steht nicht im Bericht") assert not (daten / "arcane.key").exists() @pytest.mark.parametrize("body", [b'"arc_nur_ein_text_0123456789"', b'{"schluessel": 12345678901}', b'{"schluessel": ["arc_in_einer_liste_0123456789"]}', b"kein json arc_roh_0123456789", b'{"schluessel": "arc mit leerzeichen 0123456789"}', b'{"schluessel": "kurz"}', b""]) def test_falsche_form_wiederholt_nichts(daten, arcane_nachbau, body): _, stand = arcane_nachbau antwort = _client().post("/api/homelab/einstellungen/arcane-schluessel", content=body, headers={"Content-Type": "application/json"}) assert antwort.status_code == 200 and antwort.json()["ok"] is False assert "arc_" not in antwort.text and "12345678901" not in antwort.text and "leerzeichen" not in antwort.text assert stand["aufrufe"] == [] and not (daten / "arcane.key").exists() def test_loeschen(daten): assert einstellungen.arcane_schluessel_setzen(RICHTIG)["ok"] antwort = _client().delete("/api/homelab/einstellungen/arcane-schluessel") assert antwort.json() == {"ok": True, "text": "Der Arcane-Schlüssel ist gelöscht."} assert not (daten / "arcane.key").exists() and arcane.schluessel_quelle() == "fehlt" assert _client().delete("/api/homelab/einstellungen/arcane-schluessel").json()["text"] == ( "Es war kein Arcane-Schlüssel hinterlegt.") docker = inventar._arcane_bausteine("http://127.0.0.1:9", "vm-106", {})[1] assert docker.zustand == "unbekannt" and docker.grund.startswith("Es fehlt ein Arcane-API-Schlüssel") def test_die_umgebung_geht_vor(daten, monkeypatch): assert einstellungen.arcane_schluessel_setzen(RICHTIG)["ok"] monkeypatch.setenv("MC_ARCANE_KEY", "arc_aus_der_umgebung_0123456789") assert arcane.schluessel() == "arc_aus_der_umgebung_0123456789" and arcane.schluessel_quelle() == "umgebung" assert einstellungen.stand()["arcane"]["schluessel"] == "umgebung" assert einstellungen.arcane_schluessel_setzen(FALSCH) == {"ok": False, "detail": einstellungen.UMGEBUNG_SCHLUESSEL} geloescht = einstellungen.arcane_schluessel_loeschen() assert geloescht["ok"] and "Die Umgebung setzt weiter einen (MC_ARCANE_KEY); der gilt." in geloescht["text"] assert not (daten / "arcane.key").exists() def test_echt_schalter(daten, monkeypatch): c = _client() antwort = c.post("/api/homelab/einstellungen/arcane-echt", json={"an": True}) assert antwort.status_code == 200 and antwort.json()["ok"] is True assert arcane.echt() is True assert c.get("/api/homelab/einstellungen").json()["arcane"] | {"url": ""} == { "url": "", "schluessel": "fehlt", "echt": True, "echt_quelle": "einstellung"} assert json.loads((daten / "homelab-einstellungen.json").read_text(encoding="utf-8"))["arcane"] == {"echt": True} assert c.post("/api/homelab/einstellungen/arcane-echt", json={"an": False}).json()["ok"] is True assert arcane.echt() is False and arcane.echt_quelle() == "einstellung" # Setzt die Umgebung MC_ARCANE_ECHT, gewinnt sie; der Schalter sagt es und ändert nichts. monkeypatch.setenv("MC_ARCANE_ECHT", "1") abgelehnt = c.post("/api/homelab/einstellungen/arcane-echt", json={"an": False}).json() assert abgelehnt["ok"] is False and "MC_ARCANE_ECHT" in abgelehnt["detail"] and "echt" in abgelehnt["detail"] assert arcane.echt() is True and arcane.echt_quelle() == "umgebung" monkeypatch.setenv("MC_ARCANE_ECHT", "0") assert arcane.echt() is False and einstellungen.stand()["arcane"]["echt_quelle"] == "umgebung" assert c.post("/api/homelab/einstellungen/arcane-echt", json={"an": "vielleicht"}).status_code == 422 def test_echt_schalter_macht_aus_dem_probelauf_ein_update(daten, arcane_nachbau): url, _ = arcane_nachbau assert einstellungen.arcane_schluessel_setzen(RICHTIG)["ok"] assert inventar._arcane_bausteine(url, "vm-106", {})[1].aktion.label == "Probelauf" assert einstellungen.arcane_echt_setzen(True)["ok"] assert inventar._arcane_bausteine(url, "vm-106", {})[1].aktion.label == "Jetzt updaten" # --- Herkunftsprüfung (Middleware der App, Rolle homelab) --------------------------------------------------------- _HERKUNFT = """ import json from fastapi.testclient import TestClient import app c = TestClient(app.app) fremd = {"Origin": "https://boese.example"} eigen = {"Origin": "http://testserver"} ergebnis = { "schluessel_fremd": c.post("/api/homelab/einstellungen/arcane-schluessel", json={"schluessel": "x"}, headers=fremd).status_code, "loeschen_fremd": c.delete("/api/homelab/einstellungen/arcane-schluessel", headers=fremd).status_code, "echt_fremd": c.post("/api/homelab/einstellungen/arcane-echt", json={"an": True}, headers=fremd).status_code, "alle_fremd": c.post("/api/homelab/alle", headers=fremd).status_code, "schluessel_eigen": c.post("/api/homelab/einstellungen/arcane-schluessel", json={"schluessel": "x"}, headers=eigen).json(), "lesen_fremd": c.get("/api/homelab/einstellungen", headers=fremd).status_code, } print(json.dumps(ergebnis)) """ def test_herkunftspruefung_gilt_auch_hier(tmp_path): env = {**os.environ, "MC_ROLLE": "homelab", "MC_MODELS_DIR": str(tmp_path), "MC_DATEN_DIR": str(tmp_path), "MC_PARTNER_URL": ""} env.pop("MC_ARCANE_KEY", None) lauf = subprocess.run([sys.executable, "-c", _HERKUNFT], cwd=BACKEND, env=env, capture_output=True, text=True, timeout=120) assert lauf.returncode == 0, lauf.stderr[-2000:] ergebnis = json.loads(lauf.stdout.strip().splitlines()[-1]) assert [ergebnis[k] for k in ("schluessel_fremd", "loeschen_fremd", "echt_fremd", "alle_fremd")] == [403] * 4 assert ergebnis["schluessel_eigen"]["ok"] is False # durchgelassen; nur zu kurz assert ergebnis["lesen_fremd"] == 200 assert not (tmp_path / "arcane.key").exists() and not (tmp_path / "homelab-einstellungen.json").exists()