109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
"""Regression: event factory shapes (UI relies on these field names)."""
|
|
|
|
from api.analyses.types import Finding
|
|
from api.runtime import events
|
|
from api.runtime.context import current_analysis
|
|
|
|
|
|
def test_run_start_shape():
|
|
e = events.run_start("why?")
|
|
assert e == {"type": "run_start", "question": "why?"}
|
|
|
|
|
|
def test_run_end_shape():
|
|
assert events.run_end(answer="42") == {"type": "run_end", "answer": "42", "error": None}
|
|
assert events.run_end(error="boom") == {"type": "run_end", "answer": "", "error": "boom"}
|
|
|
|
|
|
def test_plan_ready_shape():
|
|
e = events.plan_ready("rat", [{"analysis": "x", "args": {}, "why": ""}])
|
|
assert e["type"] == "plan_ready"
|
|
assert e["rationale"] == "rat"
|
|
assert isinstance(e["steps"], list)
|
|
|
|
|
|
def test_analysis_start_shape():
|
|
e = events.analysis_start("direct_answer", {"q": "?"}, "fits", step_id="direct_answer#0")
|
|
assert e == {
|
|
"type": "analysis_start",
|
|
"analysis": "direct_answer",
|
|
"step_id": "direct_answer#0",
|
|
"args": {"q": "?"},
|
|
"why": "fits",
|
|
}
|
|
|
|
|
|
def test_analysis_end_uses_finding():
|
|
f = Finding(analysis="direct_answer", summary="ok", rows=[{"a": 1}], sql=["SELECT 1"])
|
|
e = events.analysis_end("direct_answer", f, step_id="direct_answer#0")
|
|
assert e["type"] == "analysis_end"
|
|
assert e["analysis"] == "direct_answer"
|
|
assert e["step_id"] == "direct_answer#0"
|
|
assert e["summary"] == "ok"
|
|
assert e["error"] is None
|
|
assert e["row_count"] == 1
|
|
|
|
|
|
def test_analysis_fallback_shape():
|
|
e = events.analysis_fallback(
|
|
primary="drill_down", fallback="direct_answer",
|
|
reason="empty result", step_id="direct_answer#0.fallback",
|
|
)
|
|
assert e == {
|
|
"type": "analysis_fallback",
|
|
"primary": "drill_down",
|
|
"fallback": "direct_answer",
|
|
"reason": "empty result",
|
|
"step_id": "direct_answer#0.fallback",
|
|
}
|
|
|
|
|
|
def test_tool_call_shapes_carry_analysis_field():
|
|
"""Tool/LLM events must include the active Analysis name from the
|
|
contextvar — UI groups sub-events by it. None when outside an Analysis."""
|
|
# Outside any Analysis: analysis is None.
|
|
s = events.tool_call_start("compose_sql", input={"q": "?"})
|
|
assert s["type"] == "tool_call_start"
|
|
assert s["tool"] == "compose_sql"
|
|
assert s["analysis"] is None
|
|
assert s["input"] == {"q": "?"}
|
|
|
|
e = events.tool_call_end("compose_sql", output={"sql": "..."})
|
|
assert e["analysis"] is None
|
|
assert e["output"] == {"sql": "..."}
|
|
assert e["error"] is None
|
|
|
|
# Inside an Analysis context: analysis field is populated.
|
|
token = current_analysis.set("direct_answer#0")
|
|
try:
|
|
s2 = events.tool_call_start("execute_sql", input={"sql": "SELECT 1"})
|
|
e2 = events.tool_call_end("execute_sql", output={"row_count": 1})
|
|
assert s2["analysis"] == "direct_answer#0"
|
|
assert e2["analysis"] == "direct_answer#0"
|
|
finally:
|
|
current_analysis.reset(token)
|
|
|
|
|
|
def test_llm_call_carries_analysis_field():
|
|
token = current_analysis.set("compare_periods#0")
|
|
try:
|
|
e = events.llm_call("compare_periods.interpret", system_len=100, user_len=200)
|
|
assert e["type"] == "llm_call"
|
|
assert e["analysis"] == "compare_periods#0"
|
|
assert e["system_chars"] == 100
|
|
assert e["user_chars"] == 200
|
|
finally:
|
|
current_analysis.reset(token)
|
|
|
|
|
|
def test_synth_start_shape():
|
|
assert events.synth_start() == {"type": "synth_start"}
|
|
|
|
|
|
def test_sse_format_round_trip():
|
|
payload = events.tool_call_start("t", input={"x": 1})
|
|
formatted = events.sse_format(payload)
|
|
assert formatted.startswith("event: tool_call_start\n")
|
|
assert "data: " in formatted
|
|
assert formatted.endswith("\n\n")
|