111 lines
5.1 KiB
Python
111 lines
5.1 KiB
Python
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from knaster.api import create_app
|
|
|
|
from .test_router import RECHNUNG
|
|
|
|
GOOD = json.dumps({"nummer": "4711", "betrag": 99})
|
|
H = {"Authorization": "Bearer secret"}
|
|
|
|
|
|
@pytest.fixture
|
|
def client(make_system):
|
|
sysm = make_system()
|
|
sysm.chat.handlers.update({"tier1-general/rechnungen-v1": lambda m: '{"nummer": "1"}',
|
|
"tier0-strong": lambda m: GOOD})
|
|
return TestClient(create_app(sysm))
|
|
|
|
|
|
def test_auth(client):
|
|
assert client.post("/v1/route", json={"input": "x"}).status_code == 401
|
|
assert client.get("/health").status_code == 200
|
|
|
|
|
|
def test_route_and_run(client):
|
|
r = client.post("/v1/route", json={"input": RECHNUNG}, headers=H).json()
|
|
assert r["stage"] == 1 and r["adapter"] == "rechnungen-v1"
|
|
r = client.post("/v1/run", json={"input": RECHNUNG}, headers=H).json()
|
|
assert r["status"] == "ok" and r["stage"] == 0 and r["escalations"] == 1
|
|
assert r["attempts"][0]["verification"]["passed"] is False
|
|
assert client.post("/v1/route", json={"input": "x", "skill": "gibtsnicht"}, headers=H).status_code == 404
|
|
|
|
|
|
def test_metrics(client):
|
|
client.post("/v1/run", json={"input": RECHNUNG}, headers=H)
|
|
text = client.get("/metrics").text
|
|
assert 'knaster_runs_total{stage="0",status="ok"} 1.0' in text
|
|
assert 'knaster_escalations_total{signal="verifier_failed"} 1.0' in text
|
|
assert 'knaster_slot_available{slot="tier1-general"} 1.0' in text
|
|
|
|
|
|
def test_registry_and_slot_state(client):
|
|
reg = client.get("/v1/registry", headers=H).json()
|
|
assert {s["name"] for s in reg["slots"]} >= {"tier1-general", "tier0-strong"}
|
|
r = client.post("/v1/registry/slots/tier1-general/state", json={"status": "degraded", "reason": "x"},
|
|
headers=H)
|
|
assert r.json()["status"] == "degraded"
|
|
r = client.post("/v1/registry/slots/tier1-general/state", json={"status": "ok", "set_by": "n8n"}, headers=H)
|
|
assert "proposal" in r.json()
|
|
assert client.post("/v1/registry/slots/nope/state", json={"status": "ok"}, headers=H).status_code == 404
|
|
|
|
|
|
def test_maintenance_flow(client):
|
|
r = client.post("/v1/maintenance/command", json={"text": "markiere tier1-flex als degraded", "user": "robert"},
|
|
headers=H).json()
|
|
pid = r["proposals"][0]["id"]
|
|
seq = r["proposals"][0]["seq"]
|
|
open_ = client.get("/v1/maintenance/proposals", headers=H).json()
|
|
assert [p["id"] for p in open_] == [pid]
|
|
r = client.post(f"/v1/maintenance/proposals/{seq}/approve", json={"user": "robert"}, headers=H).json()
|
|
assert r["state"] == "executed"
|
|
assert client.post(f"/v1/maintenance/proposals/{seq}/approve", json={}, headers=H).status_code == 409
|
|
rep = client.post("/v1/maintenance/routine", headers=H).json()
|
|
assert "Service-Routine" in rep["text"]
|
|
|
|
|
|
def test_healer_endpoint(client):
|
|
payload = {"version": "4", "status": "firing", "alerts": [
|
|
{"status": "firing", "labels": {"alertname": "GpuTemperatureHigh", "slot": "tier1-flex"},
|
|
"fingerprint": "abc"}]}
|
|
r = client.post("/v1/healer/alertmanager", json=payload, headers=H).json()
|
|
assert r["outcomes"][0]["status"] == "executed"
|
|
reg = client.get("/v1/registry", headers=H).json()
|
|
assert next(s for s in reg["slots"] if s["name"] == "tier1-flex")["status"] == "degraded"
|
|
|
|
|
|
def test_candidate_via_api_needs_approval_then_reports(client):
|
|
r = client.post("/v1/lifecycle/candidates", json={"kind": "adapter", "slot": "tier1-general",
|
|
"target": {"cluster": "rechnungen"}}, headers=H).json()
|
|
assert r["state"] == "open"
|
|
client.post(f"/v1/maintenance/proposals/{r['seq']}/approve", json={}, headers=H)
|
|
cands = client.get("/v1/lifecycle/candidates", headers=H).json()
|
|
cid = cands[0]["id"]
|
|
assert cands[0]["state"] == "building"
|
|
r = client.post(f"/v1/lifecycle/candidates/{cid}/report", json={"step": "build", "payload": {"model": "rv2"}},
|
|
headers=H).json()
|
|
assert r["state"] == "evaluating"
|
|
assert client.post(f"/v1/lifecycle/candidates/{cid}/report", json={"step": "canary"},
|
|
headers=H).status_code == 409
|
|
|
|
|
|
def test_graduation_endpoint(client):
|
|
app_sys = client.app.state.system
|
|
app_sys.chat.handlers["tier1-general/rechnungen-v1"] = lambda m: GOOD
|
|
variants = [f"{RECHNUNG} Variante {i}" for i in range(10)]
|
|
r = client.post("/v1/skills/rechnung-extrahieren/graduate", json={"variants": variants}, headers=H).json()
|
|
assert r["report"]["graduated"] and "proposal" in r
|
|
cases = app_sys.store.list_cases()
|
|
assert len(cases) == 10 and all(c["origin"] == "graduation" for c in cases)
|
|
|
|
app_sys.chat.handlers["tier1-general/rechnungen-v1"] = lambda m: "kaputt"
|
|
r = client.post("/v1/skills/rechnung-extrahieren/graduate", json={"variants": variants}, headers=H).json()
|
|
assert not r["report"]["graduated"] and r["report"]["failures"]
|
|
assert client.post("/v1/skills/nope/graduate", json={"variants": []}, headers=H).status_code == 404
|
|
|
|
|
|
def test_reload_skills(client):
|
|
assert client.post("/v1/skills/reload", headers=H).json()["skills"] == 5
|