"""Unit tests for build_campaign_timeline (per-Run flat timeline for a campaign).""" from datetime import datetime, timedelta, timezone import pytest from agenteval.evaluation.report import build_campaign_timeline from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, EvalRun, RunStatus from agenteval.storage.repository import CampaignRepository, RunRepository from sqlmodel import Session, SQLModel, create_engine 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, []) == []