Robert RappandClaude Opus 5.5
50acfa6a95
ci / test (push) Failing after 24s
Admin-Schlüssel je Person auf ops (Bastion) aus sites/knaster/proxmox.yaml (access.ops_ssh_keys/_revoked), Playbook admins.yml, Test, Übergabe in BETRIEB §15. Einstieg für Agenten und neue Admins: AGENTS.md, Projektkarte, bekannte Probleme. Weg 2 (Verwaltung über NEX) als Entwurf in Phase 9. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
290 lines
14 KiB
Python
290 lines
14 KiB
Python
"""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].get("static_configs", []):
|
|
assert {"slot", "pve_node", "vmid"} <= set(sc["labels"]), (job, sc)
|
|
# GPU-VMs kommen per file_sd; die Labels setzt platform/proxmox/scripts/monitoring-targets.py
|
|
generated = _monitoring_targets()
|
|
for job in ("node", "gpu"):
|
|
files = [f for sd in jobs[job]["file_sd_configs"] for f in sd["files"]]
|
|
assert files == [f"/etc/knaster-sd/{job}.json"]
|
|
assert generated[job]
|
|
for entry in generated[job]:
|
|
assert {"slot", "pve_node", "vmid"} <= set(entry["labels"]), (job, entry)
|
|
from knaster.canary import OPS_LABELS
|
|
assert set(OPS_LABELS) == {"docker_host", "container", "pve_node", "vmid"}
|
|
|
|
|
|
def _targets_module():
|
|
import importlib.util
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"monitoring_targets", REPO_ROOT / "platform/proxmox/scripts/monitoring-targets.py")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
def _monitoring_targets() -> dict:
|
|
mod = _targets_module()
|
|
registry = yaml.safe_load((REPO_ROOT / "sites/knaster/registry.yaml").read_text(encoding="utf-8"))
|
|
gpu_vms = { # Form wie `tofu output -json gpu_vms`
|
|
"gpu-pve44": {"node": "pve44", "vmid": 244, "ipv4": "192.168.100.186"},
|
|
"gpu-pve48": {"node": "pve48", "vmid": 248, "ipv4": "192.168.100.183"},
|
|
"gpu-pve52": {"node": "pve52", "vmid": 252, "ipv4": "192.168.100.174"},
|
|
"gpu-pve46": {"node": "pve46", "vmid": 246, "ipv4": None},
|
|
}
|
|
return mod.build_targets(gpu_vms, registry)
|
|
|
|
|
|
def test_monitoring_targets_from_tofu_output():
|
|
t = _monitoring_targets()
|
|
assert [e["targets"][0] for e in t["node"]] == [
|
|
"192.168.100.186:9100", "192.168.100.183:9100", "192.168.100.174:9100"] # gpu-pve46 ohne IP fehlt
|
|
assert t["gpu"][0]["targets"] == ["192.168.100.186:9835"]
|
|
pve44 = t["node"][0]["labels"]
|
|
assert pve44["instance"] == "gpu-pve44" and pve44["vmid"] == "244" and pve44["pve_node"] == "pve44"
|
|
assert pve44["slot"] == "embed-default" and pve44["slots"] == "embed-default,rerank-default"
|
|
assert t["node"][2]["labels"]["slot"] == "none" # gpu-pve52 trägt noch keinen Slot
|
|
|
|
|
|
def _k3s_node(name: str, ip: str | None, ready: str = "True") -> dict:
|
|
addresses = [{"type": "Hostname", "address": name}] + ([{"type": "InternalIP", "address": ip}] if ip else [])
|
|
return {"metadata": {"name": name},
|
|
"status": {"addresses": addresses, "conditions": [{"type": "Ready", "status": ready}]}}
|
|
|
|
|
|
def test_k3s_targets_from_kubectl_nodes():
|
|
nodes = {"items": [ # Form wie `kubectl get nodes -o json`
|
|
_k3s_node("gpu-pve52", "192.168.100.174"),
|
|
_k3s_node("k3s", "192.168.100.124"),
|
|
_k3s_node("gpu-pve44", "192.168.100.186", ready="False"),
|
|
_k3s_node("gpu-pve46", None),
|
|
]}
|
|
t = _targets_module().build_k3s_targets(nodes)
|
|
assert [(e["targets"][0], e["labels"]["instance"]) for e in t["kubelet"]] == [
|
|
("192.168.100.186:10250", "gpu-pve44"), ("192.168.100.174:10250", "gpu-pve52"),
|
|
("192.168.100.124:10250", "k3s")] # gpu-pve46 ohne IP fehlt
|
|
assert t["kube-state-metrics"] == [
|
|
{"targets": ["192.168.100.124:30808"], "labels": {"instance": "kube-state-metrics"}}] # Dienst-VM bevorzugt
|
|
t = _targets_module().build_k3s_targets({"items": [nodes["items"][0], nodes["items"][2]]})
|
|
assert t["kube-state-metrics"][0]["targets"] == ["192.168.100.174:30808"] # erster bereiter Knoten
|
|
assert _targets_module().build_k3s_targets({}) == {"kubelet": [], "kube-state-metrics": []}
|
|
|
|
|
|
def test_k3s_jobs_use_generated_targets_and_nodeport():
|
|
prom = yaml.safe_load((DEPLOY / "prometheus/prometheus.yml").read_text())
|
|
jobs = {j["job_name"]: j for j in prom["scrape_configs"]}
|
|
mod = _targets_module()
|
|
written = set(mod.build_k3s_targets({}))
|
|
for job, sd in {"kubelet": "kubelet", "cadvisor": "kubelet", "kube-state-metrics": "kube-state-metrics"}.items():
|
|
assert jobs[job]["file_sd_configs"] == [{"files": [f"/etc/knaster-sd/{sd}.json"]}]
|
|
assert sd in written
|
|
for job in ("kubelet", "cadvisor"):
|
|
assert jobs[job]["scheme"] == "https" and jobs[job]["tls_config"]["ca_file"].startswith("/etc/knaster-k3s/")
|
|
docs = list(yaml.safe_load_all((DEPLOY / "k8s/monitoring/kube-state-metrics.yaml").read_text()))
|
|
svc = next(d for d in docs if d["kind"] == "Service")
|
|
assert svc["spec"]["ports"][0]["nodePort"] == mod.KSM_NODEPORT
|
|
|
|
|
|
DASHBOARDS = sorted((DEPLOY / "grafana/dashboards").glob("*.json"))
|
|
|
|
|
|
@pytest.mark.parametrize("path", DASHBOARDS, ids=[p.name for p in DASHBOARDS])
|
|
def test_grafana_dashboards_parse_with_unique_uid(path):
|
|
d = json.loads(path.read_text(encoding="utf-8"))
|
|
assert d["uid"] and d["title"] and d["panels"]
|
|
assert [x for x in DASHBOARDS if json.loads(x.read_text(encoding="utf-8"))["uid"] == d["uid"]] == [path]
|
|
|
|
|
|
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"]
|
|
if "vmid" not in ops:
|
|
continue # Ada-Slots ohne Proxmox-VM (D3); die Prometheus-Ziele folgen in Task 5.2
|
|
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["gpus"], vm # eine GPU-VM pro Knoten mit allen Karten des Knotens (D2)
|
|
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"}
|
|
gpus = [gpu for vm in vms.values() for gpu 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")
|
|
if not ops or "vmid" not in ops:
|
|
continue # Ada-Slots ohne Proxmox-VM (D3, Worker-Registry v2)
|
|
name, vm = by_vmid[ops["vmid"]]
|
|
assert vm["node"] == ops["pve_node"], slot
|
|
assert ops["docker_host"] == name, slot
|
|
# vLLM und PyTorch nur auf modernen Karten (ADR-0006)
|
|
if spec["engine"] in {"vllm", "pytorch"} and name in vms:
|
|
assert vm["driver_generation"] == "modern", 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))
|
|
|
|
|
|
def test_ops_admin_keys_in_site_file():
|
|
"""Weg 1: Admins kommen mit eigenem Schlüssel auf ops (playbooks/admins.yml)."""
|
|
access = yaml.safe_load((REPO_ROOT / "sites/knaster/proxmox.yaml").read_text())["access"]
|
|
keys, revoked = access["ops_ssh_keys"], access["ops_ssh_keys_revoked"]
|
|
assert keys # nie leer, sonst sperrt sich die Verwaltung aus
|
|
names = [k["name"] for k in keys]
|
|
assert len(names) == len(set(names))
|
|
for k in [*keys, *revoked]:
|
|
parts = k["key"].split()
|
|
assert parts[0] in {"ssh-ed25519", "ssh-rsa"} and len(parts) >= 2, k["name"]
|
|
assert "PRIVATE" not in k["key"], k["name"]
|
|
assert not {k["key"].split()[1] for k in keys} & {k["key"].split()[1] for k in revoked}
|