f8d8450b1e
「活动报告取数三件套」此前在报告/markdown/分析/对比等 7 处手写重复， 唯一深化产物 build_campaign_report_dict 被锁在周期对比私有角落。 升位为 report.py 的 load_campaign_report(session, campaign) 单一出口 （探索线 summarize_campaign_exploration 同口径），并把 8 处 scenario_names 推导式收敛为 ScenarioRepository.name_map() 窄方法。 纯结构重排、零行为变更，572 项测试全绿。
97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
"""活动报告读模型 load_campaign_report 与 ScenarioRepository.name_map 直测。
|
||
|
||
读模型是报告 / 分析 / 对比 / 导出共用的「取数 + 聚合」单一出口：
|
||
返回结果必须与手写三件套（list_by_campaign + scenario_names + generate_campaign_report）完全一致。
|
||
"""
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import pytest
|
||
from agenteval.evaluation.report import generate_campaign_report, load_campaign_report
|
||
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, Case, CaseType, EvalRun, RunStatus, Scenario
|
||
from agenteval.storage.repository import CampaignRepository, RunRepository, ScenarioRepository
|
||
from sqlmodel import Session, SQLModel, create_engine
|
||
|
||
T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
@pytest.fixture()
|
||
def db_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 / 'loader.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(session) -> Campaign:
|
||
ScenarioRepository(session).create(Scenario(
|
||
id="s-a", name="夜间问诊",
|
||
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
|
||
))
|
||
campaign = CampaignRepository(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=1)],
|
||
))
|
||
campaign.status = CampaignStatus.RUNNING
|
||
campaign.started_at = T0
|
||
campaign = CampaignRepository(session).update(campaign)
|
||
RunRepository(session).create(EvalRun(
|
||
target_id="t-1", scenario_id="s-a", campaign_id=campaign.id,
|
||
status=RunStatus.COMPLETED, started_at=T0 + timedelta(seconds=1),
|
||
summary={"total_cases": 1, "passed_cases": 1, "pass_rate": 1.0, "avg_latency_ms": 100},
|
||
))
|
||
return campaign
|
||
|
||
|
||
def test_load_campaign_report_matches_manual_trio(db_session):
|
||
campaign = _seed(db_session)
|
||
report = load_campaign_report(db_session, campaign)
|
||
|
||
runs = RunRepository(db_session).list_by_campaign(campaign.id)
|
||
scenario_names = {s.id: s.name for s in ScenarioRepository(db_session).list_all()}
|
||
expected = generate_campaign_report(campaign, runs, scenario_names=scenario_names)
|
||
|
||
assert report == expected
|
||
cap = next(c for c in report["capability_summary"] if c["scenario_id"] == "s-a")
|
||
assert cap["scenario_name"] == "夜间问诊"
|
||
|
||
|
||
def test_load_campaign_report_empty_campaign(db_session):
|
||
campaign = CampaignRepository(db_session).create(Campaign(
|
||
name="empty", target_id="t-1", window_seconds=12,
|
||
plan=[CampaignPlanEntry(scenario_id="s-x", offset_seconds=0, count=1)],
|
||
))
|
||
report = load_campaign_report(db_session, campaign)
|
||
assert report["summary"]["total_runs"] == 0
|
||
assert report["capability_summary"] == []
|
||
|
||
|
||
def test_scenario_name_map(db_session):
|
||
_seed(db_session)
|
||
ScenarioRepository(db_session).create(Scenario(
|
||
id="s-b", name="缴费引导",
|
||
cases=[Case(id="c2", type=CaseType.SINGLE, messages=["缴费"])],
|
||
))
|
||
assert ScenarioRepository(db_session).name_map() == {"s-a": "夜间问诊", "s-b": "缴费引导"}
|
||
|
||
|
||
def test_scenario_name_map_empty(db_session):
|
||
assert ScenarioRepository(db_session).name_map() == {}
|