Hitonabi/mission-control-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0190c25d5a2ea0d02cd44e1988bce4fdee49ef2f
mission-control-v2/backend/tests/test_homelab_einstellungen.py
T
HitonabiandClaude Opus 5.5 ad8dbbc1b3 Welle 1 · Live-Zahlen je Dienst auf den Kacheln, AdGuard-Zugang in den Einstellungen 
Jede Gerätekachel (Homelab-Cockpit und Übersicht) zeigt eine Live-Zahl, wo es eine gibt: Proxmox-Host „8 von 8
Gästen an“, PBS „letzte Sicherung vor 13 h“, NPMplus „10 Zertifikate gültig“ (aus der Zertifikats-Wache), Arcane
„11 von 12 Containern an“ (Arcane-API mit dem vorhandenen Schlüssel) und AdGuard „12 % geblockt“. Für AdGuard gibt es
unter Einstellungen → Homelab einen Zugang, der vor dem Speichern gegen AdGuard geprüft wird und nur im Homelab-Teil
liegt (0600, das Passwort steht in keiner Antwort). Arcane und AdGuard werden höchstens alle zwei Minuten gefragt.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 16:52:41 +02:00

262 lines
14 KiB
Python
Raw Blame History

"""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"},
"adguard": {"url": "http://192.168.178.100:8080", "zugang": "fehlt", "benutzer": None},
"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()
Reference in New Issue View Git Blame Copy Permalink