mission-control-v2/backend/tests/test_partner.py

"""Zwei Instanzen, eine Oberfläche (24.09.2026): Partner-Lebenszeichen, Durchreiche, Wächter, Rollen."""

import json
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 kern import partner
from routers import partner as partner_router
from services import waechter

BACKEND = Path(__file__).resolve().parent.parent


@pytest.fixture
def partner_server():
    """Eine Partner-Instanz zum Anfassen: /api/health antwortet, alles andere wird mitgeschrieben."""
    empfangen: list[dict] = []

    class Handler(BaseHTTPRequestHandler):
        def _antworte(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 _merke(self) -> None:
            laenge = int(self.headers.get("Content-Length") or 0)
            empfangen.append({"methode": self.command, "pfad": self.path,
                              "koerper": self.rfile.read(laenge).decode() if laenge else "",
                              "von": self.headers.get("X-MC-Von")})

        def do_GET(self):
            if self.path == "/api/health":
                self._antworte(200, {"status": "ok", "version": "t", "rolle": "homelab", "instanz": "Homelab"})
                return
            self._merke()
            self._antworte(200, {"pfad": self.path})

        def do_POST(self):
            self._merke()
            self._antworte(202, {"angenommen": True})

        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}", empfangen
    server.shutdown()


@pytest.fixture
def mit_partner(monkeypatch):
    """Setzt MC_PARTNER_URL und leert die Caches davor und danach."""
    def setzen(url: str) -> None:
        monkeypatch.setenv("MC_PARTNER_URL", url)
        einstellungen_mod.einstellungen.cache_clear()
        partner._cache.update(ts=0.0, daten=None)

    yield setzen
    monkeypatch.delenv("MC_PARTNER_URL", raising=False)
    einstellungen_mod.einstellungen.cache_clear()
    partner._cache.update(ts=0.0, daten=None)


def test_ohne_partner_nichts_eingerichtet(mit_partner):
    mit_partner("")
    assert partner.status() == {"eingerichtet": False}


def test_partner_antwortet(mit_partner, partner_server):
    url, _ = partner_server
    mit_partner(url)
    stand = partner.status()
    assert stand["erreichbar"] is True and stand["rolle"] == "homelab" and stand["name"] == "Homelab"


def test_partner_weg(mit_partner):
    mit_partner("http://127.0.0.1:9")
    stand = partner.status()
    assert stand["erreichbar"] is False and stand["fehler"] == "nicht erreichbar"


def _client() -> TestClient:
    app = FastAPI()
    app.include_router(partner_router.router)
    return TestClient(app)


def test_durchreiche_get_und_post(mit_partner, partner_server):
    url, empfangen = partner_server
    mit_partner(url)
    c = _client()
    r = c.get("/api/partner/homelab/ziele", params={"frisch": "1"})
    assert r.status_code == 200 and r.json() == {"pfad": "/api/homelab/ziele?frisch=1"}
    r = c.post("/api/partner/homelab/update", json={"ziel": "ct:101"})
    assert r.status_code == 202
    letzte = empfangen[-1]
    assert (letzte["methode"], letzte["pfad"], letzte["von"]) == ("POST", "/api/homelab/update", "Box-Wart")
    assert json.loads(letzte["koerper"]) == {"ziel": "ct:101"}


def test_instanz_ohne_netz(mit_partner):
    mit_partner("http://127.0.0.1:9")
    assert _client().get("/api/instanz").json() == {
        "rolle": "box", "instanz": "Box-Wart", "partner": {"eingerichtet": True, "name": "Homelab"}}


@pytest.mark.parametrize("pfad", ["partner/health", "partner", "stream"])
def test_durchreiche_sperrt_schleifen_und_strom(mit_partner, partner_server, pfad):
    url, empfangen = partner_server
    mit_partner(url)
    assert _client().get(f"/api/partner/{pfad}").status_code == 404
    assert empfangen == []


def test_durchreiche_ohne_partner_und_bei_ausfall(mit_partner):
    mit_partner("")
    assert _client().get("/api/partner/health").status_code == 404
    mit_partner("http://127.0.0.1:9")
    r = _client().get("/api/partner/health")
    assert r.status_code == 502 and "nicht erreichbar" in r.json()["detail"]


def test_waechter_meldet_den_partner_erst_nach_geduld(monkeypatch, tmp_path):
    """Ein Neustart der anderen Instanz (Sonntags-Update) ist kein Alarm; erst nach PARTNER_TAKTE."""
    monkeypatch.setattr(waechter, "STORE_PATH", tmp_path / "waechter.json")
    monkeypatch.setattr(waechter, "_stand", {"hinweise": {}, "kandidaten": {}, "verlauf": [], "auto": {}, "stand": 0.0})
    monkeypatch.setattr(waechter, "_update_laeuft", lambda: False)
    monkeypatch.setattr(waechter, "PRUEFUNGEN", (waechter.pruefe_partner,))
    monkeypatch.setattr(waechter, "PARTNER_TAKTE", 3)
    meldungen: list[str] = []
    monkeypatch.setattr(waechter, "_telegram", lambda betreff, text: meldungen.append(text))
    erreichbar = [False]
    monkeypatch.setattr(partner, "status", lambda frisch=False: {
        "eingerichtet": True, "name": "Homelab", "url": "http://pve:9001",
        "erreichbar": erreichbar[0], "fehler": None if erreichbar[0] else "nicht erreichbar"})

    waechter.takt()
    waechter.takt()
    assert waechter._stand["hinweise"] == {} and meldungen == []
    waechter.takt()
    assert waechter._stand["hinweise"]["partner"]["titel"] == "Homelab antwortet nicht"
    assert len(meldungen) == 1 and "Proxmox-PC (http://pve:9001) ist nicht erreichbar" in meldungen[0]

    erreichbar[0] = True
    waechter.takt()
    assert waechter._stand["hinweise"] == {}
    assert meldungen[-1] == "Erledigt: Homelab antwortet nicht."


def test_ohne_partner_kein_befund(monkeypatch):
    monkeypatch.setattr(partner, "status", lambda frisch=False: {"eingerichtet": False})
    assert waechter.pruefe_partner() == []


def _starte(rolle: str, tmp_path: Path, code: str) -> dict:
    """Die App frisch in einem eigenen Interpreter laden — die Rolle wird beim Import festgelegt."""
    env = {**os.environ, "MC_ROLLE": rolle, "MC_MODELS_DIR": str(tmp_path), "MC_DATEN_DIR": str(tmp_path),
           "MC_PARTNER_URL": ""}
    lauf = subprocess.run([sys.executable, "-c", code], cwd=BACKEND, env=env, capture_output=True, text=True,
                          timeout=120)
    assert lauf.returncode == 0, lauf.stderr[-2000:]
    return json.loads(lauf.stdout.strip().splitlines()[-1])


_PROBE = """
import json, sys
from fastapi.testclient import TestClient
import app
pfade = sorted(app.app.openapi()["paths"])
gesund = TestClient(app.app).get("/api/health").json()
box_module = [m for m in ("services.llamaswap", "services.maintenance", "routers.boxwart") if m in sys.modules]
print(json.dumps({"pfade": pfade, "gesund": gesund, "box_module": box_module}))
"""


def test_homelab_rolle_laedt_nichts_von_der_box(tmp_path):
    ergebnis = _starte("homelab", tmp_path, _PROBE)
    assert ergebnis["gesund"] == {"status": "ok", "version": ergebnis["gesund"]["version"],
                                  "rolle": "homelab", "instanz": "Homelab"}
    assert "/api/start" not in ergebnis["pfade"] and "/api/partner" in ergebnis["pfade"]
    assert ergebnis["box_module"] == []


def test_box_rolle_bleibt_wie_sie_war(tmp_path):
    ergebnis = _starte("box", tmp_path, _PROBE.replace('.get("/api/health").json()', '.get("/api/partner").json()'))
    assert {"/api/start", "/api/health", "/api/partner"} <= set(ergebnis["pfade"])
    assert ergebnis["gesund"] == {"eingerichtet": False}


def test_unbekannte_rolle_bricht_ab(monkeypatch):
    monkeypatch.setenv("MC_ROLLE", "keller")
    einstellungen_mod.einstellungen.cache_clear()
    try:
        with pytest.raises(ValueError, match="MC_ROLLE"):
            einstellungen_mod.einstellungen()
    finally:
        monkeypatch.delenv("MC_ROLLE")
        einstellungen_mod.einstellungen.cache_clear()


def test_homelab_waechter_prueft_ausfuehrer_und_gaeste(tmp_path):
    ergebnis = _starte("homelab", tmp_path, """
import json
import steward
from services import waechter
print(json.dumps({"pruefungen": [p.__name__ for p in waechter.PRUEFUNGEN], "daten": str(waechter.DATEN_DIR)}))
""")
    assert ergebnis["pruefungen"] == ["pruefe_platte", "pruefe_partner", "pruefe_ausfuehrer", "pruefe_gaeste",
                                      "pruefe_paketlisten"]


def test_homelab_rolle_hat_die_homelab_schnittstellen(tmp_path):
    ergebnis = _starte("homelab", tmp_path, _PROBE)
    assert "/api/homelab/ziele" in ergebnis["pfade"] and "/api/ziele" not in ergebnis["pfade"]