33db5aeb10
- Replay models: StepType enum, ReplayStep, ReplayPage frozen dataclasses
- Checkpoint transformer: PostgresSaver JSONB -> structured timeline steps
- Replay API: GET /api/conversations (paginated), GET /api/replay/{thread_id}
- Analytics models: AgentUsage, InterruptStats, AnalyticsResult
- Analytics event recorder: Protocol + PostgresAnalyticsRecorder + NoOp
- Analytics queries: resolution_rate, agent_usage, escalation_rate, cost, interrupts
- Analytics API: GET /api/analytics?range=Xd with envelope response
- DB migration: analytics_events table + conversations column additions
- 74 new tests, 399 total passing, 92.87% coverage
135 lines
4.2 KiB
Python
135 lines
4.2 KiB
Python
"""Unit tests for app.replay.models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
class TestStepType:
|
|
def test_all_step_types_exist(self) -> None:
|
|
from app.replay.models import StepType
|
|
|
|
assert StepType.user_message
|
|
assert StepType.supervisor_routing
|
|
assert StepType.tool_call
|
|
assert StepType.tool_result
|
|
assert StepType.agent_response
|
|
assert StepType.interrupt
|
|
|
|
def test_step_type_values(self) -> None:
|
|
from app.replay.models import StepType
|
|
|
|
assert StepType.user_message.value == "user_message"
|
|
assert StepType.tool_call.value == "tool_call"
|
|
assert StepType.agent_response.value == "agent_response"
|
|
|
|
|
|
class TestReplayStep:
|
|
def test_minimal_replay_step(self) -> None:
|
|
from app.replay.models import ReplayStep, StepType
|
|
|
|
step = ReplayStep(step=1, type=StepType.user_message, timestamp="2026-01-01T00:00:00Z")
|
|
assert step.step == 1
|
|
assert step.type == StepType.user_message
|
|
assert step.timestamp == "2026-01-01T00:00:00Z"
|
|
assert step.content == ""
|
|
assert step.agent is None
|
|
assert step.tool is None
|
|
assert step.params is None
|
|
assert step.result is None
|
|
assert step.reasoning is None
|
|
assert step.tokens is None
|
|
assert step.duration_ms is None
|
|
|
|
def test_full_replay_step(self) -> None:
|
|
from app.replay.models import ReplayStep, StepType
|
|
|
|
step = ReplayStep(
|
|
step=2,
|
|
type=StepType.tool_call,
|
|
timestamp="2026-01-01T00:00:01Z",
|
|
content="calling get_order",
|
|
agent="order_agent",
|
|
tool="get_order_status",
|
|
params={"order_id": "ORD-123"},
|
|
result={"status": "shipped"},
|
|
reasoning="user asked about order",
|
|
tokens=50,
|
|
duration_ms=200,
|
|
)
|
|
assert step.step == 2
|
|
assert step.agent == "order_agent"
|
|
assert step.tool == "get_order_status"
|
|
assert step.params == {"order_id": "ORD-123"}
|
|
assert step.tokens == 50
|
|
|
|
def test_replay_step_is_frozen(self) -> None:
|
|
from app.replay.models import ReplayStep, StepType
|
|
|
|
step = ReplayStep(step=1, type=StepType.user_message, timestamp="2026-01-01T00:00:00Z")
|
|
with pytest.raises((AttributeError, TypeError)):
|
|
step.step = 99 # type: ignore[misc]
|
|
|
|
def test_replay_step_params_is_immutable_copy(self) -> None:
|
|
from app.replay.models import ReplayStep, StepType
|
|
|
|
params = {"key": "value"}
|
|
step = ReplayStep(
|
|
step=1,
|
|
type=StepType.tool_call,
|
|
timestamp="2026-01-01T00:00:00Z",
|
|
params=params,
|
|
)
|
|
# Modifying original dict should not affect step
|
|
params["new_key"] = "new_value"
|
|
assert "new_key" not in (step.params or {})
|
|
|
|
|
|
class TestReplayPage:
|
|
def test_replay_page_construction(self) -> None:
|
|
from app.replay.models import ReplayPage, ReplayStep, StepType
|
|
|
|
steps = (
|
|
ReplayStep(step=1, type=StepType.user_message, timestamp="2026-01-01T00:00:00Z"),
|
|
ReplayStep(step=2, type=StepType.agent_response, timestamp="2026-01-01T00:00:01Z"),
|
|
)
|
|
page = ReplayPage(
|
|
thread_id="thread-123",
|
|
total_steps=2,
|
|
page=1,
|
|
per_page=20,
|
|
steps=steps,
|
|
)
|
|
assert page.thread_id == "thread-123"
|
|
assert page.total_steps == 2
|
|
assert page.page == 1
|
|
assert page.per_page == 20
|
|
assert len(page.steps) == 2
|
|
|
|
def test_replay_page_is_frozen(self) -> None:
|
|
from app.replay.models import ReplayPage
|
|
|
|
page = ReplayPage(
|
|
thread_id="t1",
|
|
total_steps=0,
|
|
page=1,
|
|
per_page=20,
|
|
steps=(),
|
|
)
|
|
with pytest.raises((AttributeError, TypeError)):
|
|
page.page = 2 # type: ignore[misc]
|
|
|
|
def test_replay_page_empty_steps(self) -> None:
|
|
from app.replay.models import ReplayPage
|
|
|
|
page = ReplayPage(
|
|
thread_id="t1",
|
|
total_steps=0,
|
|
page=1,
|
|
per_page=20,
|
|
steps=(),
|
|
)
|
|
assert page.steps == ()
|