100 lines
3.5 KiB
Python
100 lines
3.5 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 export_quality_snapshot as snapshot_script # noqa: E402
|
|
import report_failed_turns as failed_turns_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_report_failed_turns_writes_output_file(tmp_path: Path) -> None:
|
|
output = tmp_path / "failed-turns.json"
|
|
payload = {"records": 3, "failed_turns": []}
|
|
failed_turns_script.write_report(output, payload)
|
|
assert json.loads(output.read_text(encoding="utf-8"))["records"] == 3
|
|
|
|
|
|
def test_export_quality_snapshot_writes_dated_artifacts_and_index(tmp_path: Path) -> None:
|
|
root = tmp_path / "reports" / "observability" / "onec-agent"
|
|
write_jsonl(
|
|
root / "turn_audit" / "20260705.jsonl",
|
|
[
|
|
{
|
|
"event_type": "turn_audit",
|
|
"timestamp": "2026-07-02T10:00:00+00:00",
|
|
"turn_id": "bad-1",
|
|
"request_id": "req-1",
|
|
"outcome": "failure",
|
|
"failure_type": "adapter_error",
|
|
"error_code": "adapter_error",
|
|
"error_message": "missing base_id",
|
|
"user_text": "Проверь сохраненное состояние",
|
|
},
|
|
{
|
|
"event_type": "turn_audit",
|
|
"timestamp": "2026-07-03T10:00:00+00:00",
|
|
"turn_id": "bad-2",
|
|
"request_id": "req-2",
|
|
"outcome": "failure",
|
|
"failure_type": "adapter_error",
|
|
"error_code": "adapter_error",
|
|
"error_message": "missing base_id",
|
|
"user_text": "Проверь сохраненное состояние",
|
|
},
|
|
],
|
|
)
|
|
write_jsonl(
|
|
root / "model_calls" / "20260705.jsonl",
|
|
[
|
|
{"event_type": "model_calls", "route_name": "1c", "served_model_name": "stub-model", "latency_ms": 20},
|
|
],
|
|
)
|
|
write_jsonl(
|
|
root / "tool_calls" / "20260705.jsonl",
|
|
[
|
|
{"event_type": "tool_calls", "tool_name": "tool.echo", "duration_ms": 5},
|
|
],
|
|
)
|
|
write_jsonl(
|
|
root / "retrieval_events" / "20260705.jsonl",
|
|
[
|
|
{"event_type": "retrieval_events", "plugin": "1c", "profile": "official", "sources_json": [{"a": 1}]},
|
|
],
|
|
)
|
|
|
|
output_dir = tmp_path / "reports" / "observability" / "quality"
|
|
payload = snapshot_script.build_snapshot(
|
|
report_dir=tmp_path / "reports" / "observability",
|
|
output_dir=output_dir,
|
|
days=0,
|
|
limit=10,
|
|
min_repeat_count=2,
|
|
failed_limit=100,
|
|
)
|
|
|
|
quality_path = Path(payload["paths"]["quality"])
|
|
repeated_path = Path(payload["paths"]["repeated_failures"])
|
|
failed_path = Path(payload["paths"]["failed_turns"])
|
|
index_path = Path(payload["paths"]["index"])
|
|
|
|
assert quality_path.exists()
|
|
assert repeated_path.exists()
|
|
assert failed_path.exists()
|
|
assert index_path.exists()
|
|
|
|
index = json.loads(index_path.read_text(encoding="utf-8"))
|
|
assert index["snapshots"]
|
|
assert index["snapshots"][0]["summary"]["failed_turns"] == 2
|
|
assert index["snapshots"][0]["summary"]["repeated_failures"] == 1
|