cluster-knaster/core/tests/test_artifacts.py

"""Strukturprüfungen der Deployment-Artefakte (ohne laufende Infrastruktur)."""

import json

import httpx
import pytest
import yaml

from knaster.healer.engine import HealerConfig
from knaster.settings import REPO_ROOT

DEPLOY = REPO_ROOT / "deploy"


def test_alert_names_match_healer_rules():
    alerts = yaml.safe_load((DEPLOY / "prometheus/alerts.yml").read_text())
    names = {r["alert"] for g in alerts["groups"] for r in g["rules"]}
    rules = HealerConfig.from_file(REPO_ROOT / "config/healer-rules.yaml").rules
    matched = {r.match["alertname"] for r in rules}
    assert names == matched


def test_healer_labels_available_on_alerts():
    """Labels, die Healer-Regeln brauchen, müssen von Metrik oder Scrape-Target kommen."""
    prom = yaml.safe_load((DEPLOY / "prometheus/prometheus.yml").read_text())
    jobs = {j["job_name"]: j for j in prom["scrape_configs"]}
    for job in ("node", "gpu", "llamacpp"):
        for sc in jobs[job]["static_configs"]:
            assert {"slot", "pve_node", "vmid"} <= set(sc["labels"]), (job, sc)
    from knaster.canary import OPS_LABELS
    assert set(OPS_LABELS) == {"docker_host", "container", "pve_node", "vmid"}


def test_alertmanager_targets_router_endpoint():
    am = yaml.safe_load((DEPLOY / "alertmanager/alertmanager.yml").read_text())
    url = am["receivers"][0]["webhook_configs"][0]["url"]
    assert url.endswith("/v1/healer/alertmanager")
    assert am["receivers"][0]["webhook_configs"][0]["send_resolved"] is True


@pytest.mark.parametrize("role", ["infra", "gpu-vllm", "gpu-llamacpp"])
def test_compose_files_parse(role):
    data = yaml.safe_load((DEPLOY / f"compose/{role}/docker-compose.yml").read_text())
    assert data["services"]
    assert (DEPLOY / f"compose/{role}/.env.example").exists()


def test_ops_labels_consistent_between_registry_and_prometheus():
    """Das erste Scrape-Ziel eines Slots trägt dieselben ops-Labels wie die Registry (Healer-Ziel)."""
    reg = yaml.safe_load((REPO_ROOT / "sites/knaster/registry.yaml").read_text())
    prom = yaml.safe_load((DEPLOY / "prometheus/prometheus.yml").read_text())
    first: dict[str, dict] = {}
    for sc in next(j for j in prom["scrape_configs"] if j["job_name"] == "llamacpp")["static_configs"]:
        first.setdefault(sc["labels"]["slot"], sc["labels"])
    assert first
    for slot, labels in first.items():
        ops = reg["slots"][slot]["ops"]
        assert {k: ops[k] for k in ("docker_host", "container", "pve_node", "vmid")} == \
            {k: labels[k] for k in ("docker_host", "container", "pve_node", "vmid")}, slot


WORKFLOWS = sorted((REPO_ROOT / "n8n/workflows").glob("*.json"))


@pytest.mark.parametrize("path", WORKFLOWS, ids=[p.name for p in WORKFLOWS])
def test_n8n_workflow_structure(path):
    wf = json.loads(path.read_text())
    assert wf["active"] is False
    names = [n["name"] for n in wf["nodes"]]
    assert len(names) == len(set(names))
    ids = [n["id"] for n in wf["nodes"]]
    assert len(ids) == len(set(ids))
    for src, outs in wf["connections"].items():
        assert src in names
        for branch in outs["main"]:
            for target in branch:
                assert target["node"] in names
    # jeder Knoten außer dem Trigger ist erreichbar
    targets = {t["node"] for o in wf["connections"].values() for b in o["main"] for t in b}
    triggers = [n for n in wf["nodes"] if "trigger" in n["type"].lower() or n["type"].endswith(".webhook")]
    assert len(triggers) == 1
    assert set(names) - targets == {triggers[0]["name"]}


def test_n8n_router_paths_exist(make_system):
    from knaster.api import create_app

    paths = {r.path for r in create_app(make_system()).routes}
    for path in WORKFLOWS:
        for n in json.loads(path.read_text())["nodes"]:
            url = n["parameters"].get("url", "")
            if "router_url" in url:
                assert url.split("}}")[-1] in paths


def test_webhook_notifier_sends_auth_header():
    from knaster.healer.drivers import WebhookNotifier

    seen = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["auth"] = request.headers.get("authorization")
        seen["body"] = json.loads(request.content)
        return httpx.Response(200)

    n = WebhookNotifier("http://n8n/webhook/knaster/notify", "Bearer t",
                        client=httpx.Client(transport=httpx.MockTransport(handler)))
    n.notify("hallo", "high")
    assert seen == {"auth": "Bearer t", "body": {"text": "hallo", "severity": "high", "data": {}}}


def test_drivers_build_correct_requests():
    from knaster.healer.drivers import PortainerDriver, ProxmoxDriver

    calls = []

    def handler(request: httpx.Request) -> httpx.Response:
        calls.append((request.method, request.url.path, dict(request.headers)))
        if request.url.path == "/api/endpoints":
            return httpx.Response(200, json=[{"Id": 3, "Name": "gpu-main"}])
        return httpx.Response(200, json={"data": "UPID:1"})

    client = httpx.Client(transport=httpx.MockTransport(handler))
    PortainerDriver("https://portainer", "k", client=client).restart_container("gpu-main", "vllm")
    assert calls[-1][:2] == ("POST", "/api/endpoints/3/docker/containers/vllm/restart")
    assert calls[-1][2]["x-api-key"] == "k"
    ProxmoxDriver("https://pve:8006", "knaster@pve!healer", "s", client=client).restart_vm("pve-ada", "201")
    assert calls[-1][:2] == ("POST", "/api2/json/nodes/pve-ada/qemu/201/status/reboot")
    assert calls[-1][2]["authorization"] == "PVEAPIToken=knaster@pve!healer=s"


def test_llm_and_embedding_clients():
    from knaster.embeddings import LlamaCppReranker, OpenAIEmbedder
    from knaster.llm import ContextOverflowError, EndpointError, OpenAIChatClient

    def handler(request: httpx.Request) -> httpx.Response:
        body = json.loads(request.content)
        if request.url.path == "/v1/embeddings":
            return httpx.Response(200, json={"data": [{"index": 1, "embedding": [0.0, 1.0]},
                                                      {"index": 0, "embedding": [1.0, 0.0]}]})
        if request.url.path == "/v1/rerank":
            return httpx.Response(200, json={"results": [{"index": 1, "relevance_score": 0.9},
                                                         {"index": 0, "relevance_score": 0.1}]})
        if body["model"] == "overflow":
            return httpx.Response(400, text="This model's maximum context length is 8192 tokens")
        if body["model"] == "down":
            return httpx.Response(503, text="unavailable")
        return httpx.Response(200, json={"model": body["model"], "usage": {"total_tokens": 3},
                                         "choices": [{"message": {"content": "hi"}, "finish_reason": "stop"}]})

    client = httpx.Client(transport=httpx.MockTransport(handler))
    assert OpenAIEmbedder("http://x", "e", client=client).embed(["a", "b"]) == [[1.0, 0.0], [0.0, 1.0]]
    assert LlamaCppReranker("http://x", client=client).rerank("q", ["a", "b"]) == [0.1, 0.9]
    chat = OpenAIChatClient("http://x", "k", client=client)
    assert chat.chat("m", [{"role": "user", "content": "x"}]).content == "hi"
    with pytest.raises(ContextOverflowError):
        chat.chat("overflow", [])
    with pytest.raises(EndpointError):
        chat.chat("down", [])


def test_proxmox_site_consistent_with_registry_and_prometheus():
    site = yaml.safe_load((REPO_ROOT / "sites/knaster/proxmox.yaml").read_text())
    reg = yaml.safe_load((REPO_ROOT / "sites/knaster/registry.yaml").read_text())
    vms = site["vms"]
    for vm in vms.values():
        for gpu in vm["gpus"]:
            assert gpu in site["pci_mappings"]
            assert site["pci_mappings"][gpu]["node"] == vm["node"]  # Mapping liegt auf dem Host der VM
        assert vm["driver_generation"] in {"legacy", "modern"}
        assert vm["node"] in site["nodes"] and site["nodes"][vm["node"]]["gpu_host"]
    gpus = [g for vm in vms.values() for g in vm["gpus"]]
    assert len(gpus) == len(set(gpus))  # keine GPU doppelt durchgereicht
    assert set(gpus) == set(site["pci_mappings"])  # jede gemappte Karte steckt in einer VM
    by_vmid = {str(vm["vmid"]): (name, vm) for name, vm in {**vms, **site["containers"]}.items()}
    for slot, spec in reg["slots"].items():
        ops = spec.get("ops") or {}
        if "vmid" not in ops:  # z. B. Ada-Worker unter Windows (D3), kein Proxmox-Gast
            assert "pve_node" not in ops, slot
            continue
        name, vm = by_vmid[ops["vmid"]]
        assert vm["node"] == ops["pve_node"], slot
        assert ops["docker_host"] == name, slot
    for c in site["containers"].values():
        if c.get("ha"):
            assert site["nodes"][c["node"]]["zfs_tank"] and c["storage"] == "tank"
            assert all(site["nodes"][t]["zfs_tank"] for t in c["replicate_to"])
    vmids = [str(v["vmid"]) for v in vms.values()] + [str(c["vmid"]) for c in site["containers"].values()]
    assert len(vmids) == len(set(vmids))
    # vLLM nur auf modernen Karten
    for vm in vms.values():
        if vm["role"] == "gpu-vllm":
            assert vm["driver_generation"] == "modern"