205 lines
8.5 KiB
Python
205 lines
8.5 KiB
Python
import http.server
|
|
import json
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from ai_incidents import judge
|
|
from ai_incidents.judge import JudgeConfig, JudgeError, build_message, claude_argv, extract_json, parse_verdict
|
|
from ai_incidents.ledger import Entry
|
|
from ai_incidents.prefilter import Candidate
|
|
from ai_incidents.redact import Redactor
|
|
from helpers import judge_cmd
|
|
|
|
|
|
def cands(n=3):
|
|
out = []
|
|
for i in range(1, n + 1):
|
|
c = Candidate("SNIPPET", "claude-code", f"sess{i:04d}", "2026-07-1" + str(i), f" [user] thing {i}")
|
|
c.id = f"C{i:02d}"
|
|
out.append(c)
|
|
return out
|
|
|
|
|
|
GOOD = {
|
|
"incidents": [{
|
|
"title": "Wiped the cache", "date": "2026-07-12", "severity": "high", "category": "data-loss",
|
|
"what": "w", "cost": "c", "lesson": "l", "candidates": ["C01", "C99"], "why": "because",
|
|
}],
|
|
"excluded": [{"candidates": ["C02"], "reason": "routine", "duplicate_of": "E01"}],
|
|
"patterns": [],
|
|
}
|
|
|
|
|
|
def test_extract_json_tolerates_fences_and_prose():
|
|
assert extract_json('Sure!\n```json\n{"a": 1}\n```\nthanks') == {"a": 1}
|
|
assert extract_json('verdict: {"a": {"b": 2}} trailing') == {"a": {"b": 2}}
|
|
with pytest.raises(JudgeError):
|
|
extract_json("You've hit your monthly spend limit.")
|
|
with pytest.raises(JudgeError):
|
|
extract_json("{not json")
|
|
|
|
|
|
def test_parse_verdict_validates_and_normalises():
|
|
v = parse_verdict(json.dumps(GOOD), cands(), "2026-09-21")
|
|
[i] = v.incidents
|
|
assert i.severity == "HIGH" and i.candidates == ["C01"] # unknown id dropped
|
|
assert v.excluded[0].duplicate_of == "E01"
|
|
assert v.unaddressed == ["C03"]
|
|
|
|
|
|
def test_malformed_incidents_are_rejected_not_filed():
|
|
bad = {"incidents": [{"title": "x", "severity": "CATASTROPHIC", "what": "w", "cost": "c", "lesson": "l"},
|
|
{"title": "no lesson", "severity": "LOW", "what": "w", "cost": "c"},
|
|
"not an object"],
|
|
"excluded": []}
|
|
v = parse_verdict(json.dumps(bad), cands(), "2026-09-21")
|
|
assert v.incidents == [] and len(v.rejected) == 3
|
|
|
|
|
|
def test_bad_date_falls_back_to_candidate_date():
|
|
raw = {"incidents": [dict(GOOD["incidents"][0], date="last Tuesday", candidates=["C02"])], "excluded": []}
|
|
assert parse_verdict(json.dumps(raw), cands(), "2026-09-21").incidents[0].date == "2026-07-12"
|
|
|
|
|
|
def test_unknown_category_becomes_other():
|
|
raw = {"incidents": [dict(GOOD["incidents"][0], category="vibes")], "excluded": []}
|
|
assert parse_verdict(json.dumps(raw), cands(), "2026-09-21").incidents[0].category == "other"
|
|
|
|
|
|
def test_verdict_text_is_redacted():
|
|
raw = {"incidents": [dict(GOOD["incidents"][0], what="printed password=abcdef123456 to the log")],
|
|
"excluded": []}
|
|
v = parse_verdict(json.dumps(raw), cands(), "2026-09-21", Redactor())
|
|
assert "abcdef123456" not in v.incidents[0].what
|
|
|
|
|
|
def test_non_object_answers_fail():
|
|
with pytest.raises(JudgeError):
|
|
parse_verdict('{"incidents": "none"}', cands(), "2026-09-21")
|
|
with pytest.raises(JudgeError):
|
|
parse_verdict('{"summary": "all quiet"}', cands(), "2026-09-21")
|
|
|
|
|
|
def test_build_message_lists_candidates_and_ledger():
|
|
msg = build_message(cands(2), [Entry.new("Old one", "2026-07-01", "LOW", "what it did", "c", "l")], "2026-09-21")
|
|
assert "[C01] SNIPPET · claude-code · sess0001" in msg
|
|
assert "[E01] Old one · 2026-07-01 · LOW -- what it did" in msg
|
|
|
|
|
|
def test_claude_argv_disables_every_tool_and_keeps_content_off_argv():
|
|
argv = claude_argv(JudgeConfig(model="sonnet"), "RUBRIC")
|
|
assert argv[:2] == ["claude", "-p"]
|
|
i = argv.index("--tools")
|
|
assert argv[i + 1] == ""
|
|
for flag in ("--strict-mcp-config", "--no-session-persistence"):
|
|
assert flag in argv
|
|
assert "--mcp-config" not in argv
|
|
assert argv[argv.index("--model") + 1] == "sonnet"
|
|
|
|
|
|
def _fake_claude(tmp_path, envelope: dict | str, code: int = 0) -> str:
|
|
"""A stand-in `claude` binary that also records its argv and stdin."""
|
|
p = tmp_path / "claude"
|
|
out = envelope if isinstance(envelope, str) else json.dumps(envelope)
|
|
p.write_text(
|
|
"#!/usr/bin/env python3\nimport sys, json, os\n"
|
|
f"open({str(tmp_path / 'argv.json')!r}, 'w').write(json.dumps(sys.argv[1:]))\n"
|
|
f"open({str(tmp_path / 'stdin.txt')!r}, 'w').write(sys.stdin.read())\n"
|
|
f"open({str(tmp_path / 'cwd.txt')!r}, 'w').write(os.getcwd())\n"
|
|
f"sys.stdout.write({out!r})\nsys.exit({code})\n"
|
|
)
|
|
p.chmod(0o755)
|
|
return str(p)
|
|
|
|
|
|
def test_claude_backend_reads_envelope_and_usage(tmp_path):
|
|
env = {"type": "result", "subtype": "success", "is_error": False, "result": json.dumps(GOOD),
|
|
"total_cost_usd": 0.12, "usage": {"input_tokens": 10, "cache_read_input_tokens": 5, "output_tokens": 7}}
|
|
cfg = JudgeConfig(binary=_fake_claude(tmp_path, env))
|
|
text, usage = judge.call(cfg, "RUBRIC", "CANDIDATES GO HERE")
|
|
assert json.loads(text) == GOOD
|
|
assert usage == {"total_cost_usd": 0.12, "input_tokens": 15, "output_tokens": 7}
|
|
assert (tmp_path / "stdin.txt").read_text() == "CANDIDATES GO HERE"
|
|
assert "CANDIDATES GO HERE" not in (tmp_path / "argv.json").read_text()
|
|
assert "ai-incidents-judge-" in (tmp_path / "cwd.txt").read_text()
|
|
|
|
|
|
def test_claude_backend_errors(tmp_path):
|
|
cfg = JudgeConfig(binary=_fake_claude(tmp_path, {"is_error": True, "result": "You've hit your spend limit"}))
|
|
with pytest.raises(JudgeError, match="spend limit"):
|
|
judge.call(cfg, "R", "M")
|
|
cfg = JudgeConfig(binary=_fake_claude(tmp_path, "plain text, no envelope"))
|
|
with pytest.raises(JudgeError, match="envelope"):
|
|
judge.call(cfg, "R", "M")
|
|
cfg = JudgeConfig(binary=_fake_claude(tmp_path, "", code=1))
|
|
with pytest.raises(JudgeError, match="exited 1"):
|
|
judge.call(cfg, "R", "M")
|
|
with pytest.raises(JudgeError, match="not found"):
|
|
judge.call(JudgeConfig(binary=str(tmp_path / "missing")), "R", "M")
|
|
|
|
|
|
def test_command_backend(monkeypatch):
|
|
monkeypatch.setenv("FAKE_JUDGE_MODE", "file-first")
|
|
cfg = JudgeConfig(backend="command", command=judge_cmd())
|
|
text, _ = judge.call(cfg, "RUBRIC", "[C01] SNIPPET · x\n[C02] SNIPPET · y")
|
|
v = parse_verdict(text, cands(2), "2026-09-21")
|
|
assert [i.candidates for i in v.incidents] == [["C01"]] and v.excluded[0].candidates == ["C02"]
|
|
|
|
|
|
def test_command_backend_timeout(tmp_path):
|
|
cfg = JudgeConfig(backend="command", command=["sleep", "5"], timeout=1)
|
|
with pytest.raises(JudgeError, match="timed out"):
|
|
judge.call(cfg, "R", "M")
|
|
|
|
|
|
class _Handler(http.server.BaseHTTPRequestHandler):
|
|
seen = {}
|
|
|
|
def do_POST(self):
|
|
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
_Handler.seen = {"path": self.path, "body": body, "auth": self.headers.get("Authorization")}
|
|
resp = {"choices": [{"message": {"content": json.dumps(GOOD)}}],
|
|
"usage": {"prompt_tokens": 100, "completion_tokens": 20}}
|
|
data = json.dumps(resp).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
|
|
def test_openai_backend(monkeypatch):
|
|
srv = http.server.HTTPServer(("127.0.0.1", 0), _Handler)
|
|
t = threading.Thread(target=srv.serve_forever, daemon=True)
|
|
t.start()
|
|
try:
|
|
monkeypatch.setenv("LOCAL_LLM_KEY", "k-123")
|
|
cfg = JudgeConfig(backend="openai", model="qwen3", base_url=f"http://127.0.0.1:{srv.server_port}/v1",
|
|
api_key_env="LOCAL_LLM_KEY")
|
|
text, usage = judge.call(cfg, "RUBRIC", "MSG")
|
|
finally:
|
|
srv.shutdown()
|
|
assert json.loads(text) == GOOD
|
|
assert usage == {"input_tokens": 100, "output_tokens": 20}
|
|
s = _Handler.seen
|
|
assert s["path"] == "/v1/chat/completions" and s["auth"] == "Bearer k-123"
|
|
assert s["body"]["messages"][0] == {"role": "system", "content": "RUBRIC"}
|
|
assert s["body"]["response_format"] == {"type": "json_object"} and s["body"]["temperature"] == 0
|
|
|
|
|
|
def test_openai_backend_missing_key_and_unreachable(monkeypatch):
|
|
monkeypatch.delenv("NOPE_KEY", raising=False)
|
|
with pytest.raises(JudgeError, match="NOPE_KEY"):
|
|
judge.call(JudgeConfig(backend="openai", api_key_env="NOPE_KEY"), "R", "M")
|
|
with pytest.raises(JudgeError, match="unreachable"):
|
|
judge.call(JudgeConfig(backend="openai", base_url="http://127.0.0.1:9/v1", timeout=2), "R", "M")
|
|
|
|
|
|
def test_default_rubric_ships_with_the_package():
|
|
r = judge.default_rubric()
|
|
assert "Answer format" in r and "misattribution" in r.lower()
|