126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
"""Pure contract tests for intelligent-evaluation read projections."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from agenteval.intelligent_eval.models import (
|
|
IntelligentEval,
|
|
IntelligentEvalSession,
|
|
IntelligentEvalSessionStatus,
|
|
IntelligentEvalStatus,
|
|
)
|
|
from agenteval.intelligent_eval.read_model import IntelligentEvalReadModel, project_detail, project_list_item
|
|
from agenteval.intelligent_eval.repository import IntelligentEvalRepository, IntelligentEvalSessionRepository
|
|
from sqlalchemy import event
|
|
from sqlmodel import Session
|
|
|
|
|
|
def _evaluation(eval_id: str = "eval-1") -> IntelligentEval:
|
|
now = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
return IntelligentEval(
|
|
id=eval_id,
|
|
name="客服评估",
|
|
target_id="target-1",
|
|
status=IntelligentEvalStatus.EXECUTING,
|
|
goal="验证退货流程",
|
|
seeds={"personas": ["老客户"]},
|
|
intent="流程覆盖",
|
|
role_description="模拟用户",
|
|
plan={"dimensions": ["退货"]},
|
|
time_window_hours=24,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
|
|
def _session(
|
|
session_id: str,
|
|
status: IntelligentEvalSessionStatus,
|
|
eval_id: str = "eval-1",
|
|
) -> IntelligentEvalSession:
|
|
return IntelligentEvalSession(
|
|
id=session_id,
|
|
eval_id=eval_id,
|
|
target_id="target-1",
|
|
persona={"name": session_id},
|
|
goal="完成退货",
|
|
dimension="退货",
|
|
status=status,
|
|
turn_count=2,
|
|
)
|
|
|
|
|
|
def test_list_projection_contains_stable_fields_and_counts() -> None:
|
|
projection = project_list_item(
|
|
_evaluation(),
|
|
[_session("s-1", IntelligentEvalSessionStatus.RUNNING), _session("s-2", IntelligentEvalSessionStatus.COMPLETED)],
|
|
)
|
|
|
|
assert projection.id == "eval-1"
|
|
assert projection.session_count == 2
|
|
assert projection.completed_sessions == 1
|
|
assert projection.plan == {"dimensions": ["退货"]}
|
|
|
|
|
|
def test_detail_projection_contains_session_metadata_without_messages() -> None:
|
|
projection = project_detail(_evaluation(), [_session("s-1", IntelligentEvalSessionStatus.COMPLETED)])
|
|
payload = projection.model_dump(mode="json")
|
|
|
|
assert payload["sessions"][0]["id"] == "s-1"
|
|
assert payload["sessions"][0]["turn_count"] == 2
|
|
assert "messages" not in payload
|
|
assert "content" not in payload["sessions"][0]
|
|
|
|
|
|
def _count_queries(session: Session, action) -> int:
|
|
count = 0
|
|
|
|
def before_cursor_execute(*_args) -> None:
|
|
nonlocal count
|
|
count += 1
|
|
|
|
engine = session.get_bind()
|
|
event.listen(engine, "before_cursor_execute", before_cursor_execute)
|
|
try:
|
|
action()
|
|
finally:
|
|
event.remove(engine, "before_cursor_execute", before_cursor_execute)
|
|
return count
|
|
|
|
|
|
def test_empty_list_projection_does_not_query_sessions(db_session: Session) -> None:
|
|
reader = IntelligentEvalReadModel(db_session)
|
|
|
|
query_count = _count_queries(db_session, lambda: reader.list_items([]))
|
|
|
|
assert query_count == 0
|
|
|
|
|
|
def test_list_projection_loads_all_sessions_with_one_query(db_session: Session) -> None:
|
|
sessions = IntelligentEvalSessionRepository(db_session)
|
|
sessions._create(_session("s-1", IntelligentEvalSessionStatus.RUNNING, "eval-1"))
|
|
sessions._create(_session("s-2", IntelligentEvalSessionStatus.COMPLETED, "eval-2"))
|
|
reader = IntelligentEvalReadModel(db_session)
|
|
evaluations = [_evaluation("eval-1"), _evaluation("eval-2")]
|
|
result: list = []
|
|
|
|
query_count = _count_queries(db_session, lambda: result.extend(reader.list_items(evaluations)))
|
|
|
|
assert query_count == 1
|
|
assert [(item.id, item.session_count) for item in result] == [("eval-1", 1), ("eval-2", 1)]
|
|
|
|
|
|
def test_detail_projection_uses_one_snapshot_query(db_session: Session) -> None:
|
|
evaluations = IntelligentEvalRepository(db_session)
|
|
sessions = IntelligentEvalSessionRepository(db_session)
|
|
evaluations._create(_evaluation())
|
|
sessions._create(_session("s-1", IntelligentEvalSessionStatus.COMPLETED))
|
|
reader = IntelligentEvalReadModel(db_session)
|
|
result: list = []
|
|
|
|
query_count = _count_queries(db_session, lambda: result.append(reader.detail_by_id("eval-1")))
|
|
|
|
assert query_count == 1
|
|
assert result[0] is not None
|
|
assert result[0].id == "eval-1"
|
|
assert [item.id for item in result[0].sessions] == ["s-1"]
|