62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
import summarize_observability_reports as summary_script # noqa: E402
|
|
|
|
|
|
def write_jsonl(path: Path, records: list[dict]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8") as handle:
|
|
for record in records:
|
|
handle.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
|
|
|
|
def test_summarize_observability_reports_groups_records(tmp_path: Path) -> None:
|
|
root = tmp_path / "reports" / "observability" / "onec-agent"
|
|
write_jsonl(
|
|
root / "access_events" / "20260705.jsonl",
|
|
[
|
|
{"event_type": "access_events", "status_code": 200},
|
|
{"event_type": "access_events", "status_code": 400},
|
|
],
|
|
)
|
|
write_jsonl(
|
|
root / "turn_audit" / "20260705.jsonl",
|
|
[
|
|
{"event_type": "turn_audit", "plugin": "1c", "outcome": "success", "failure_type": "none"},
|
|
{"event_type": "turn_audit", "plugin": "1c", "outcome": "failure", "failure_type": "adapter_error"},
|
|
],
|
|
)
|
|
write_jsonl(
|
|
root / "model_calls" / "20260705.jsonl",
|
|
[
|
|
{"event_type": "model_calls", "route_name": "1c", "latency_ms": 10},
|
|
{"event_type": "model_calls", "route_name": "1c", "latency_ms": 20},
|
|
],
|
|
)
|
|
write_jsonl(
|
|
root / "tool_calls" / "20260705.jsonl",
|
|
[
|
|
{"event_type": "tool_calls", "tool_name": "metadata.saved_state.forms.search", "duration_ms": 7},
|
|
],
|
|
)
|
|
|
|
records = summary_script.iter_records(tmp_path / "reports" / "observability")
|
|
summary = summary_script.summarize(records)
|
|
|
|
assert summary["records"] == 7
|
|
assert summary["by_event_type"]["access_events"] == 2
|
|
assert summary["http_status_counts"]["200"] == 1
|
|
assert summary["turns_by_plugin"]["1c"]["success"] == 1
|
|
assert summary["turns_by_plugin"]["1c"]["failure"] == 1
|
|
assert summary["failure_type_counts"]["adapter_error"] == 1
|
|
assert summary["model_routes"][0]["route_name"] == "1c"
|
|
assert summary["model_routes"][0]["avg_latency_ms"] == 15
|
|
assert summary["tool_calls"][0]["tool_name"] == "metadata.saved_state.forms.search"
|