AgentEvalTool/tests/unit/test_campaign_report.py

"""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, summarize_campaign_progress
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


def test_summarize_campaign_progress_uses_planned_total_and_pass_rate(report_session):
    # plan totals 2 + 1 = 3 planned runs; only two have completed so far.
    campaign = CampaignRepository(report_session).create(Campaign(
        name="cycle", target_id="t-1", window_seconds=12, time_scale=1.0,
        plan=[
            CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=2),
            CampaignPlanEntry(scenario_id="s-b", offset_seconds=6, count=1),
        ],
    ))
    campaign.status = CampaignStatus.RUNNING
    campaign.started_at = T0
    campaign = CampaignRepository(report_session).update(campaign)

    _seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0, pass_rate=1.0)
    _seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 1, pass_rate=0.0)
    _seed_child(report_session, campaign.id, "s-b", RunStatus.FAILED, 6)  # execution failure

    runs = RunRepository(report_session).list_by_campaign(campaign.id)
    progress = summarize_campaign_progress(campaign, runs)

    assert progress["planned_total"] == 3          # Σ plan.count, not spawned-so-far
    assert progress["completed_runs"] == 2          # FAILED does not count as completed
    # ADR-0002: failed execution counts as 0.0 → (1.0 + 0.0 + 0.0) / 3
    assert progress["overall_pass_rate"] == round(1.0 / 3, 4)