File size: 2,274 Bytes
0366d65 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """Agent-trace export + redaction (server/trace.py) and the UI download handler."""
import json
from server import events as bus
from server import trace
from ui.blocks import _on_analyze, _on_download_trace
def _drive_a_run():
# stub mode is forced by conftest -> this produces a real run on the bus
list(_on_analyze("Mom: dinner Sunday 6pm", None, None))
def test_export_run_shape_after_analyze():
_drive_a_run()
t = trace.export_run()
assert t["schema"] == "imessage-cal-trace"
assert t["version"] == 1
assert t["run_id"]
assert t["steps"], "expected at least one step from the run"
assert t["summary"]["events"] >= 1
allowed = {
"stage", "level", "ts", "latency_ms", "events",
"conflicts", "images", "tokens", "message",
}
for step in t["steps"]:
assert set(step).issubset(allowed), f"unexpected keys: {set(step) - allowed}"
def test_redaction_strips_chat_names():
with bus.run_scope("analyze"):
bus.emit("ingest", "2 msg(s) from 3rd grade chat", images=0)
bus.emit("decision", "1 event(s) detected", events=1)
rid = trace.export_run()["run_id"]
redacted = trace.export_run(run_id=rid, redact=True)
assert all("3rd grade chat" not in s["message"] for s in redacted["steps"])
assert redacted["redacted"] is True
full = trace.export_run(run_id=rid, redact=False)
assert any("3rd grade chat" in s["message"] for s in full["steps"])
def test_empty_envelope_when_no_run():
# a run id that doesn't exist -> valid but empty
t = trace.export_run(run_id="nonexistent")
assert t["steps"] == []
assert t["summary"]["events"] == 0
def test_write_trace_roundtrips(tmp_path):
_drive_a_run()
t = trace.export_run()
path = trace.write_trace(t, path=str(tmp_path / "t.json"))
with open(path, encoding="utf-8") as f:
assert json.load(f) == t
def test_download_handler_paths():
_drive_a_run()
path, msg = _on_download_trace(True)
assert path and "step" in msg
bus._BUF.clear() # simulate a fresh process: no runs to export
none_path, none_msg = _on_download_trace(True)
assert none_path is None and "No agent run" in none_msg
|