8bc5aa6979
build_campaign_timeline flattens a campaign's child Runs into offset-sorted
per-Run entries (distinct from the report's 12-bucket aggregation), reusing a
shared _run_window_offset口径 so both views place a run identically. Exposes
GET /campaigns/{id}/timeline and the api.ts type/call. (v0.6 ticket 06)
157 lines
6.5 KiB
Python
157 lines
6.5 KiB
Python
"""Unit tests for build_campaign_timeline (per-Run flat timeline for a campaign)."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
from agenteval.evaluation.report import build_campaign_timeline
|
|
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, EvalRun, RunStatus
|
|
from agenteval.storage.repository import CampaignRepository, RunRepository
|
|
|
|
T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
@pytest.fixture()
|
|
def timeline_session(tmp_path):
|
|
from agenteval.storage.db import ( # noqa: F401
|
|
CampaignDB, EvalResultDB, EvalRunDB, EvalTargetDB, FileCategoryDB,
|
|
FileRecordDB, ScenarioDB, TurnDB,
|
|
)
|
|
engine = create_engine(
|
|
f"sqlite:///{tmp_path / 'campaign_timeline.db'}",
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
SQLModel.metadata.create_all(engine)
|
|
session = Session(engine)
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def _seed_campaign(session, *, window_seconds=12, time_scale=1.0) -> Campaign:
|
|
campaign = CampaignRepository(session).create(Campaign(
|
|
name="cycle", target_id="t-1", window_seconds=window_seconds, time_scale=time_scale,
|
|
plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
|
|
))
|
|
campaign.status = CampaignStatus.RUNNING
|
|
campaign.started_at = T0
|
|
return CampaignRepository(session).update(campaign)
|
|
|
|
|
|
def _seed_child(session, campaign_id, scenario_id, status, offset, *, pass_rate=None, latency=None) -> EvalRun:
|
|
summary = None
|
|
if status == RunStatus.COMPLETED:
|
|
summary = {
|
|
"total_cases": 1,
|
|
"passed_cases": 1 if (pass_rate or 0) >= 1 else 0,
|
|
"pass_rate": pass_rate,
|
|
"avg_latency_ms": latency,
|
|
}
|
|
return RunRepository(session).create(EvalRun(
|
|
target_id="t-1", scenario_id=scenario_id, campaign_id=campaign_id,
|
|
status=status, started_at=T0 + timedelta(seconds=offset) if offset is not None else None,
|
|
summary=summary,
|
|
))
|
|
|
|
|
|
def _entry(timeline, run_id):
|
|
return next(e for e in timeline if e["run_id"] == run_id)
|
|
|
|
|
|
def test_offset_status_and_fields(timeline_session):
|
|
campaign = _seed_campaign(timeline_session)
|
|
r0 = _seed_child(timeline_session, campaign.id, "s-a", RunStatus.COMPLETED, 0, pass_rate=1.0, latency=100)
|
|
r6 = _seed_child(timeline_session, campaign.id, "s-b", RunStatus.COMPLETED, 6, pass_rate=0.5, latency=300)
|
|
|
|
timeline = build_campaign_timeline(
|
|
campaign, RunRepository(timeline_session).list_by_campaign(campaign.id),
|
|
scenario_names={"s-a": "夜间问诊", "s-b": "白班"},
|
|
)
|
|
|
|
assert len(timeline) == 2
|
|
e0 = _entry(timeline, r0.id)
|
|
assert e0["scenario_id"] == "s-a"
|
|
assert e0["scenario_name"] == "夜间问诊"
|
|
assert e0["offset_seconds"] == 0.0
|
|
assert e0["status"] == "completed"
|
|
assert e0["pass_rate"] == 1.0
|
|
assert e0["avg_latency_ms"] == 100
|
|
assert e0["started_at"] is not None
|
|
|
|
e6 = _entry(timeline, r6.id)
|
|
assert e6["offset_seconds"] == 6.0
|
|
assert e6["scenario_name"] == "白班"
|
|
assert e6["pass_rate"] == 0.5
|
|
|
|
|
|
def test_time_scale_scales_offset_only(timeline_session):
|
|
# window 12s, scale 10 → a run 0.6s in maps to offset 6.0; numbers untouched.
|
|
campaign = _seed_campaign(timeline_session, window_seconds=12, time_scale=10.0)
|
|
r = _seed_child(timeline_session, campaign.id, "s-a", RunStatus.COMPLETED, 0.6, pass_rate=0.8, latency=120)
|
|
timeline = build_campaign_timeline(campaign, RunRepository(timeline_session).list_by_campaign(campaign.id))
|
|
e = _entry(timeline, r.id)
|
|
assert e["offset_seconds"] == 6.0
|
|
assert e["pass_rate"] == 0.8
|
|
assert e["avg_latency_ms"] == 120
|
|
|
|
|
|
def test_offset_clamped_to_window(timeline_session):
|
|
campaign = _seed_campaign(timeline_session, window_seconds=12)
|
|
r = _seed_child(timeline_session, campaign.id, "s-a", RunStatus.COMPLETED, 999, pass_rate=1.0, latency=50)
|
|
timeline = build_campaign_timeline(campaign, RunRepository(timeline_session).list_by_campaign(campaign.id))
|
|
assert _entry(timeline, r.id)["offset_seconds"] == 12.0
|
|
|
|
|
|
def test_pending_run_field_mapping(timeline_session):
|
|
campaign = _seed_campaign(timeline_session)
|
|
r = _seed_child(timeline_session, campaign.id, "s-a", RunStatus.PENDING, 0)
|
|
timeline = build_campaign_timeline(campaign, RunRepository(timeline_session).list_by_campaign(campaign.id))
|
|
e = _entry(timeline, r.id)
|
|
assert e["offset_seconds"] == 0.0
|
|
assert e["status"] == "pending"
|
|
assert e["pass_rate"] is None
|
|
assert e["avg_latency_ms"] is None
|
|
|
|
|
|
def test_missing_started_at_defaults_offset_zero(timeline_session):
|
|
# Defensive: a Run with no started_at (never persisted) sits at offset 0.
|
|
campaign = _seed_campaign(timeline_session)
|
|
run = EvalRun(target_id="t-1", scenario_id="s-a", campaign_id=campaign.id,
|
|
status=RunStatus.PENDING, started_at=None)
|
|
timeline = build_campaign_timeline(campaign, [run])
|
|
assert timeline[0]["offset_seconds"] == 0.0
|
|
assert timeline[0]["started_at"] is None
|
|
|
|
|
|
def test_failed_run_has_no_summary_numbers(timeline_session):
|
|
campaign = _seed_campaign(timeline_session)
|
|
r = _seed_child(timeline_session, campaign.id, "s-a", RunStatus.FAILED, 6)
|
|
timeline = build_campaign_timeline(campaign, RunRepository(timeline_session).list_by_campaign(campaign.id))
|
|
e = _entry(timeline, r.id)
|
|
assert e["status"] == "failed"
|
|
assert e["pass_rate"] is None
|
|
assert e["avg_latency_ms"] is None
|
|
|
|
|
|
def test_entries_sorted_by_offset(timeline_session):
|
|
campaign = _seed_campaign(timeline_session)
|
|
late = _seed_child(timeline_session, campaign.id, "s-a", RunStatus.COMPLETED, 9, pass_rate=1.0, latency=10)
|
|
early = _seed_child(timeline_session, campaign.id, "s-b", RunStatus.COMPLETED, 1, pass_rate=1.0, latency=10)
|
|
timeline = build_campaign_timeline(campaign, RunRepository(timeline_session).list_by_campaign(campaign.id))
|
|
assert [e["run_id"] for e in timeline] == [early.id, late.id]
|
|
|
|
|
|
def test_scenario_name_falls_back_to_id_prefix(timeline_session):
|
|
campaign = _seed_campaign(timeline_session)
|
|
_seed_child(timeline_session, campaign.id, "scenario-xyz-123", RunStatus.COMPLETED, 0, pass_rate=1.0, latency=10)
|
|
timeline = build_campaign_timeline(campaign, RunRepository(timeline_session).list_by_campaign(campaign.id))
|
|
assert timeline[0]["scenario_name"] == "scenario"[:8]
|
|
|
|
|
|
def test_empty_campaign(timeline_session):
|
|
campaign = _seed_campaign(timeline_session)
|
|
assert build_campaign_timeline(campaign, []) == []
|