f433ebb970
Add generate_campaign_report: a pure aggregator over a campaign's child Runs
producing a time-trend axis (Runs bucketed by service-window position) and a
capability-summary axis (grouped by scenario), each carrying pass_rate /
availability / latency. pass_rate keeps the single-Run case-level meaning and
counts execution failures as 0.0 (ADR-0002); time_scale only places Runs into
window-time buckets and never alters any figure. Engine summary now records
avg_latency_ms to feed the latency axis.
Expose GET /api/campaigns/{id}/report (structured) and .../report/markdown
(reusing the existing Markdown export path). Adds "可用性/Availability" to the
domain glossary.
154 lines
6.5 KiB
Python
154 lines
6.5 KiB
Python
"""Unit tests for generate_campaign_report (dual-axis campaign report)."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
from agenteval.evaluation.report import generate_campaign_report
|
|
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 report_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_report.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), summary=summary,
|
|
))
|
|
|
|
|
|
def _bucket(report, idx):
|
|
return report["time_trend"][idx]
|
|
|
|
|
|
def _cap(report, sid):
|
|
return next(c for c in report["capability_summary"] if c["scenario_id"] == sid)
|
|
|
|
|
|
# ── happy path: multi-bucket, multi-scenario, with a failure ────────────────
|
|
|
|
def test_dual_axis_values(report_session):
|
|
campaign = _seed_campaign(report_session) # window 12s, 12 buckets → 1s each
|
|
_seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0, pass_rate=1.0, latency=100)
|
|
_seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0, pass_rate=0.0, latency=200)
|
|
_seed_child(report_session, campaign.id, "s-b", RunStatus.COMPLETED, 6, pass_rate=0.5, latency=300)
|
|
_seed_child(report_session, campaign.id, "s-b", RunStatus.FAILED, 6)
|
|
|
|
report = generate_campaign_report(campaign, RunRepository(report_session).list_by_campaign(campaign.id))
|
|
|
|
assert len(report["time_trend"]) == 12
|
|
|
|
b0 = _bucket(report, 0)
|
|
assert b0["run_count"] == 2
|
|
assert b0["pass_rate"] == 0.5 # (1.0 + 0.0) / 2
|
|
assert b0["availability"] == 1.0
|
|
assert b0["avg_latency_ms"] == 150.0 # (100 + 200) / 2
|
|
|
|
b6 = _bucket(report, 6)
|
|
assert b6["run_count"] == 2
|
|
assert b6["pass_rate"] == 0.25 # (0.5 + 0[failed]) / 2 — failure counts (ADR-0002)
|
|
assert b6["availability"] == 0.5 # 1 of 2 completed
|
|
assert b6["avg_latency_ms"] == 300.0 # failed run has no latency
|
|
|
|
# empty bucket
|
|
assert _bucket(report, 3)["run_count"] == 0
|
|
assert _bucket(report, 3)["pass_rate"] is None
|
|
|
|
cap_a = _cap(report, "s-a")
|
|
assert cap_a["run_count"] == 2 and cap_a["pass_rate"] == 0.5 and cap_a["avg_latency_ms"] == 150.0
|
|
cap_b = _cap(report, "s-b")
|
|
assert cap_b["run_count"] == 2 and cap_b["pass_rate"] == 0.25 and cap_b["availability"] == 0.5
|
|
|
|
s = report["summary"]
|
|
assert s["total_runs"] == 4
|
|
assert s["completed_runs"] == 3
|
|
assert s["overall_pass_rate"] == 0.375 # (1 + 0 + 0.5 + 0) / 4
|
|
assert s["overall_availability"] == 0.75
|
|
assert s["avg_latency_ms"] == 200.0 # (100 + 200 + 300) / 3
|
|
|
|
|
|
def test_scenario_names_mapping(report_session):
|
|
campaign = _seed_campaign(report_session)
|
|
_seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0, pass_rate=1.0, latency=100)
|
|
report = generate_campaign_report(
|
|
campaign, RunRepository(report_session).list_by_campaign(campaign.id),
|
|
scenario_names={"s-a": "夜间问诊"},
|
|
)
|
|
assert _cap(report, "s-a")["scenario_name"] == "夜间问诊"
|
|
|
|
|
|
# ── boundaries: empty / partial ─────────────────────────────────────────────
|
|
|
|
def test_empty_campaign_no_division_by_zero(report_session):
|
|
campaign = _seed_campaign(report_session)
|
|
report = generate_campaign_report(campaign, [])
|
|
assert len(report["time_trend"]) == 12
|
|
assert all(b["run_count"] == 0 and b["pass_rate"] is None for b in report["time_trend"])
|
|
assert report["capability_summary"] == []
|
|
assert report["summary"]["total_runs"] == 0
|
|
assert report["summary"]["overall_pass_rate"] is None
|
|
assert report["summary"]["avg_latency_ms"] is None
|
|
|
|
|
|
def test_only_in_flight_runs_partial_aggregate(report_session):
|
|
campaign = _seed_campaign(report_session)
|
|
_seed_child(report_session, campaign.id, "s-a", RunStatus.RUNNING, 0)
|
|
_seed_child(report_session, campaign.id, "s-a", RunStatus.PENDING, 0)
|
|
report = generate_campaign_report(campaign, RunRepository(report_session).list_by_campaign(campaign.id))
|
|
s = report["summary"]
|
|
assert s["total_runs"] == 2
|
|
assert s["completed_runs"] == 0
|
|
assert s["overall_pass_rate"] == 0.0 # nothing completed yet, failures/incomplete count as 0
|
|
assert s["overall_availability"] == 0.0
|
|
assert s["avg_latency_ms"] is None
|
|
|
|
|
|
def test_time_scale_only_affects_bucketing_not_numbers(report_session):
|
|
# Compressed campaign: window 12s, scale 10 → a run 0.6s in maps to offset 6.
|
|
campaign = _seed_campaign(report_session, window_seconds=12, time_scale=10.0)
|
|
_seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0.6, pass_rate=0.8, latency=120)
|
|
report = generate_campaign_report(campaign, RunRepository(report_session).list_by_campaign(campaign.id))
|
|
# Lands in bucket 6 (0.6s * 10 = 6.0), and the numbers are untouched by scale.
|
|
assert _bucket(report, 6)["run_count"] == 1
|
|
assert _bucket(report, 6)["pass_rate"] == 0.8
|
|
assert _bucket(report, 6)["avg_latency_ms"] == 120.0
|