9cf64ab0e2
Adds unit tests for the campaign report beyond value correctness: cancellation exclusion from denominators (ADR-0004), window clamping, naive/aware timestamp equivalence, plus schema validation of the report dict (key sets, types, rate bounds, contiguous buckets, capability sort, ISO-UTC timestamps) and consistency with the single aggregate_runs seam.
236 lines
9.2 KiB
Python
236 lines
9.2 KiB
Python
"""Report aggregation logic & output format validation (报告聚合与格式校验).
|
||
|
||
补充 test_campaign_report.py（数值口径）与 test_metrics.py（单点聚合）之外：
|
||
- 聚合边界：用户取消不计分母（ADR-0004）、窗口外/窗口前运行钳制、
|
||
SQLite naive 时间戳与 aware 等价；
|
||
- 格式校验：报告 dict 的键集合、类型、取值范围、时间序列连续性、
|
||
能力面排序与 ISO-UTC 时间戳格式。generate_campaign_report 是纯函数，
|
||
直接构造模型对象，不落库。
|
||
"""
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from agenteval.evaluation.metrics import aggregate_runs
|
||
from agenteval.evaluation.report import generate_campaign_report
|
||
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, EvalRun, RunStatus
|
||
|
||
T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
def _campaign(*, window_seconds: int = 12, time_scale: float = 1.0, started_at=T0, completed_at=None) -> Campaign:
|
||
return Campaign(
|
||
id="camp-fmt",
|
||
name="格式校验活动",
|
||
target_id="t-1",
|
||
window_seconds=window_seconds,
|
||
time_scale=time_scale,
|
||
plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
|
||
status=CampaignStatus.RUNNING,
|
||
started_at=started_at,
|
||
completed_at=completed_at,
|
||
)
|
||
|
||
|
||
def _run(
|
||
scenario_id: str = "s-a",
|
||
*,
|
||
status: RunStatus = RunStatus.COMPLETED,
|
||
offset_seconds: float = 0.0,
|
||
pass_rate: float | None = None,
|
||
latency: float | None = None,
|
||
error: dict | None = None,
|
||
started_at: datetime | None = None,
|
||
) -> EvalRun:
|
||
summary = None
|
||
if pass_rate is not None or latency is not None or error is not None:
|
||
summary = {}
|
||
if pass_rate is not None:
|
||
summary["pass_rate"] = pass_rate
|
||
if latency is not None:
|
||
summary["avg_latency_ms"] = latency
|
||
if error is not None:
|
||
summary["error"] = error
|
||
return EvalRun(
|
||
target_id="t-1",
|
||
scenario_id=scenario_id,
|
||
campaign_id="camp-fmt",
|
||
status=status,
|
||
started_at=started_at if started_at is not None else T0 + timedelta(seconds=offset_seconds),
|
||
summary=summary,
|
||
)
|
||
|
||
|
||
def _report(runs, **kwargs):
|
||
return generate_campaign_report(_campaign(), runs, **kwargs)
|
||
|
||
|
||
# ── 格式校验：顶层结构 ────────────────────────────────────────────────
|
||
|
||
|
||
def test_top_level_schema_keys_and_types():
|
||
report = _report([_run(pass_rate=1.0, latency=100)])
|
||
assert set(report) == {
|
||
"campaign_id",
|
||
"name",
|
||
"target_id",
|
||
"status",
|
||
"window_seconds",
|
||
"time_scale",
|
||
"started_at",
|
||
"completed_at",
|
||
"summary",
|
||
"time_trend",
|
||
"capability_summary",
|
||
}
|
||
assert report["campaign_id"] == "camp-fmt"
|
||
assert isinstance(report["window_seconds"], int) and report["window_seconds"] > 0
|
||
assert isinstance(report["time_scale"], float)
|
||
assert report["status"] == CampaignStatus.RUNNING.value
|
||
assert isinstance(report["time_trend"], list)
|
||
assert isinstance(report["capability_summary"], list)
|
||
|
||
|
||
def test_summary_schema_types_and_bounds():
|
||
report = _report(
|
||
[
|
||
_run(pass_rate=1.0, latency=100),
|
||
_run(scenario_id="s-b", status=RunStatus.FAILED),
|
||
]
|
||
)
|
||
summary = report["summary"]
|
||
assert set(summary) == {
|
||
"total_runs",
|
||
"completed_runs",
|
||
"overall_pass_rate",
|
||
"overall_availability",
|
||
"avg_latency_ms",
|
||
}
|
||
assert isinstance(summary["total_runs"], int) and summary["total_runs"] == 2
|
||
assert isinstance(summary["completed_runs"], int) and summary["completed_runs"] == 1
|
||
for key in ("overall_pass_rate", "overall_availability"):
|
||
value = summary[key]
|
||
assert value is None or (isinstance(value, float) and 0.0 <= value <= 1.0), key
|
||
assert summary["avg_latency_ms"] is None or summary["avg_latency_ms"] >= 0
|
||
|
||
|
||
def test_summary_matches_single_aggregation_seam():
|
||
"""summary 数值必须与单点聚合口径 aggregate_runs 完全一致（ADR-0004）。"""
|
||
runs = [
|
||
_run(pass_rate=1.0, latency=100),
|
||
_run(pass_rate=0.5, latency=300),
|
||
_run(status=RunStatus.FAILED),
|
||
]
|
||
report = _report(runs)
|
||
agg = aggregate_runs(runs)
|
||
summary = report["summary"]
|
||
assert summary["overall_pass_rate"] == agg["pass_rate"]
|
||
assert summary["overall_availability"] == agg["availability"]
|
||
assert summary["avg_latency_ms"] == agg["avg_latency_ms"]
|
||
|
||
|
||
# ── 格式校验：时间趋势序列 ────────────────────────────────────────────
|
||
|
||
|
||
def test_time_trend_contiguous_buckets_with_custom_count():
|
||
report = generate_campaign_report(_campaign(window_seconds=12), [_run(pass_rate=1.0)], bucket_count=4)
|
||
trend = report["time_trend"]
|
||
assert len(trend) == 4
|
||
for idx, bucket in enumerate(trend):
|
||
assert bucket["bucket_index"] == idx
|
||
assert bucket["start_seconds"] == idx * 3.0 # window / bucket_count
|
||
assert bucket["end_seconds"] == (idx + 1) * 3.0
|
||
assert {"run_count", "pass_rate", "availability", "avg_latency_ms"} <= set(bucket)
|
||
# 相邻桶首尾相接
|
||
for prev, cur in zip(trend, trend[1:]):
|
||
assert prev["end_seconds"] == cur["start_seconds"]
|
||
|
||
|
||
def test_time_trend_bucket_types_and_bounds():
|
||
report = _report([_run(pass_rate=0.8, latency=120)])
|
||
for bucket in report["time_trend"]:
|
||
assert isinstance(bucket["run_count"], int) and bucket["run_count"] >= 0
|
||
if bucket["run_count"] == 0:
|
||
assert bucket["pass_rate"] is None and bucket["availability"] is None
|
||
else:
|
||
assert 0.0 <= bucket["pass_rate"] <= 1.0
|
||
assert 0.0 <= bucket["availability"] <= 1.0
|
||
|
||
|
||
# ── 格式校验：能力面汇总 ──────────────────────────────────────────────
|
||
|
||
|
||
def test_capability_summary_schema_and_sorted_by_run_count_desc():
|
||
runs = [
|
||
_run("s-a", pass_rate=1.0),
|
||
_run("s-a", pass_rate=0.5),
|
||
_run("s-a", pass_rate=0.0),
|
||
_run("s-b", pass_rate=1.0),
|
||
]
|
||
report = _report(runs, scenario_names={"s-a": "夜间问诊"})
|
||
caps = report["capability_summary"]
|
||
assert [c["scenario_id"] for c in caps] == ["s-a", "s-b"] # run_count 降序
|
||
for cap in caps:
|
||
assert {"scenario_id", "scenario_name", "run_count", "pass_rate", "availability", "avg_latency_ms"} <= set(cap)
|
||
assert cap["run_count"] >= 1
|
||
assert caps[0]["scenario_name"] == "夜间问诊"
|
||
assert caps[1]["scenario_name"] == "s-b"[:8] # 无映射时回落 id 前缀
|
||
|
||
|
||
# ── 格式校验：时间戳 ISO-UTC ──────────────────────────────────────────
|
||
|
||
|
||
def test_timestamps_serialize_as_iso_utc_or_none():
|
||
report = generate_campaign_report(_campaign(completed_at=T0 + timedelta(hours=1)), [_run()])
|
||
assert report["started_at"].endswith("Z")
|
||
assert report["completed_at"].endswith("Z")
|
||
assert "T" in report["started_at"]
|
||
|
||
unfinished = generate_campaign_report(_campaign(), [_run()])
|
||
assert unfinished["completed_at"] is None
|
||
|
||
|
||
def test_naive_campaign_timestamps_still_serialize_utc():
|
||
naive_start = datetime(2026, 1, 1, 0, 0, 0) # SQLite round-trip drops tzinfo
|
||
report = generate_campaign_report(_campaign(started_at=naive_start), [_run()])
|
||
assert report["started_at"].endswith("Z")
|
||
|
||
|
||
# ── 聚合边界：取消 / 钳制 / naive 时间 ────────────────────────────────
|
||
|
||
|
||
def test_cancelled_run_excluded_from_summary_but_counted():
|
||
runs = [
|
||
_run(pass_rate=1.0, latency=100),
|
||
_run(status=RunStatus.FAILED, error={"code": "cancelled_by_user", "message": "stop"}),
|
||
]
|
||
summary = _report(runs)["summary"]
|
||
assert summary["total_runs"] == 2 # 发生过的事仍然可见
|
||
assert summary["completed_runs"] == 1
|
||
assert summary["overall_pass_rate"] == 1.0 # 取消不进分母（ADR-0004）
|
||
assert summary["overall_availability"] == 1.0
|
||
|
||
|
||
def test_run_started_before_window_clamped_to_first_bucket():
|
||
report = _report([_run(pass_rate=1.0, started_at=T0 - timedelta(seconds=10))])
|
||
assert report["time_trend"][0]["run_count"] == 1
|
||
|
||
|
||
def test_run_started_after_window_clamped_to_last_bucket():
|
||
report = _report([_run(pass_rate=1.0, started_at=T0 + timedelta(seconds=100))])
|
||
assert report["time_trend"][-1]["run_count"] == 1
|
||
assert sum(b["run_count"] for b in report["time_trend"]) == 1
|
||
|
||
|
||
def test_naive_run_timestamps_bucket_same_as_aware():
|
||
aware = _run(pass_rate=1.0, started_at=T0 + timedelta(seconds=6))
|
||
naive = _run(pass_rate=1.0, started_at=datetime(2026, 1, 1, 0, 0, 6)) # SQLite 读回的 naive UTC
|
||
trend_aware = _report([aware])["time_trend"]
|
||
trend_naive = _report([naive])["time_trend"]
|
||
assert trend_aware == trend_naive
|
||
assert trend_naive[6]["run_count"] == 1 # window 12s / 12 buckets → offset 6 落第 6 桶
|
||
|
||
|
||
def test_unstarted_run_sits_at_offset_zero():
|
||
report = _report([_run(pass_rate=1.0, started_at=None)])
|
||
assert report["time_trend"][0]["run_count"] == 1
|