"""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", "vllm"): 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(): reg = yaml.safe_load((REPO_ROOT / "sites/knaster/registry.yaml").read_text()) prom = yaml.safe_load((DEPLOY / "prometheus/prometheus.yml").read_text()) vllm = {sc["labels"]["slot"]: sc["labels"] for sc in next(j for j in prom["scrape_configs"] if j["job_name"] == "vllm")["static_configs"]} for slot, labels in vllm.items(): ops = reg["slots"][slot]["ops"] assert ops["vmid"] == labels["vmid"] and ops["container"] == labels["container"] 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(): assert vm["gpu"] in site["pci_mappings"] assert site["pci_mappings"][vm["gpu"]]["node"] == vm["node"] # Mapping liegt auf dem Host der VM assert vm["driver_generation"] in {"legacy", "modern"} gpus = [vm["gpu"] for vm in vms.values()] assert len(gpus) == len(set(gpus)) # keine GPU doppelt durchgereicht by_vmid = {str(vm["vmid"]): (name, vm) for name, vm in vms.items()} for slot, spec in reg["slots"].items(): ops = spec.get("ops") if not ops: continue name, vm = by_vmid[ops["vmid"]] assert vm["node"] == ops["pve_node"], slot assert ops["docker_host"].startswith(name.split("-")[0]), slot 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"