8a526599ab
报告层推导连通用例标记（无判定结果 + 每轮有回复 + 无用例级错误）， summary 新增 connectivity_cases 与 judged_pass_rate（无判定型用例时为 null）。 对比报告同步标注且连通用例按引擎口径计通过；总通过率口径不变（ADR-0002）。
344 lines
13 KiB
Python
344 lines
13 KiB
Python
"""Unit tests for report generation: generate_report, generate_compare_report, render_markdown_report."""
|
||
|
||
import pytest
|
||
from sqlmodel import Session, SQLModel, create_engine
|
||
|
||
from agenteval.evaluation.report import (
|
||
generate_compare_report,
|
||
generate_report,
|
||
render_markdown_report,
|
||
render_json_report,
|
||
)
|
||
from agenteval.models import (
|
||
Case, CaseType, EvalResult, EvalRun, EvalTarget, RunStatus, Scenario, Turn,
|
||
PlatformType, ChannelType, TargetStatus,
|
||
)
|
||
from agenteval.storage.repository import ResultRepository, RunRepository, ScenarioRepository, TargetRepository
|
||
|
||
|
||
@pytest.fixture()
|
||
def report_session(tmp_path):
|
||
from agenteval.storage.db import ( # noqa: F401
|
||
EvalResultDB, EvalRunDB, EvalTargetDB, FileCategoryDB, FileRecordDB, ScenarioDB, TurnDB,
|
||
)
|
||
engine = create_engine(
|
||
f"sqlite:///{tmp_path / 'report_test.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_run(
|
||
session: Session,
|
||
*,
|
||
pass_rate: float = 1.0,
|
||
n_cases: int = 1,
|
||
scenario_id: str | None = None,
|
||
connectivity_cases: int = 0,
|
||
errored_cases: int = 0,
|
||
) -> str:
|
||
"""Create a minimal completed run with real data in the DB and return run_id.
|
||
|
||
connectivity_cases: extra cases with replied turns but no rule results.
|
||
errored_cases: extra cases with a missing-reply turn and no rule results.
|
||
"""
|
||
target = EvalTarget(
|
||
name="测试对象",
|
||
platform=PlatformType.AI_DIGITAL_EMPLOYEE,
|
||
channel_type=ChannelType.TUTU_API,
|
||
channel_config={},
|
||
status=TargetStatus.ACTIVE,
|
||
)
|
||
target = TargetRepository(session).create(target)
|
||
|
||
if scenario_id is None:
|
||
scenario = Scenario(
|
||
name="测试场景",
|
||
cases=(
|
||
[Case(id=f"c{i}", type=CaseType.SINGLE, messages=["hi"]) for i in range(n_cases)]
|
||
+ [Case(id=f"conn{i}", type=CaseType.SINGLE, messages=["ping"]) for i in range(connectivity_cases)]
|
||
+ [Case(id=f"err{i}", type=CaseType.SINGLE, messages=["ping"]) for i in range(errored_cases)]
|
||
),
|
||
)
|
||
scenario = ScenarioRepository(session).create(scenario)
|
||
scenario_id = scenario.id
|
||
|
||
run = EvalRun(
|
||
target_id=target.id,
|
||
scenario_id=scenario_id,
|
||
status=RunStatus.COMPLETED,
|
||
)
|
||
run = RunRepository(session).create(run)
|
||
|
||
run_repo = RunRepository(session)
|
||
result_repo = ResultRepository(session)
|
||
|
||
total = n_cases
|
||
passed = int(total * pass_rate)
|
||
|
||
for i in range(n_cases):
|
||
turn = Turn(
|
||
run_id=run.id,
|
||
case_id=f"c{i}",
|
||
round_index=1,
|
||
sent_message={"msgBody": {"content": f"问题{i}"}},
|
||
reply={"msgBody": {"content": f"回答{i}"}},
|
||
latency_ms=200,
|
||
)
|
||
result_repo.save_turn(turn)
|
||
db_turn = run_repo.get_turns(run.id)[-1]
|
||
|
||
eval_result = EvalResult(
|
||
run_id=run.id,
|
||
case_id=f"c{i}",
|
||
turn_id=db_turn.id or "",
|
||
rule_type="keyword_match",
|
||
passed=(i < passed),
|
||
score=1.0 if i < passed else 0.0,
|
||
reason="通过" if i < passed else "失败",
|
||
)
|
||
result_repo.save_result(eval_result)
|
||
|
||
for i in range(connectivity_cases):
|
||
result_repo.save_turn(Turn(
|
||
run_id=run.id,
|
||
case_id=f"conn{i}",
|
||
round_index=1,
|
||
sent_message={"msgBody": {"content": "ping"}},
|
||
reply={"msgBody": {"content": "pong"}},
|
||
latency_ms=100,
|
||
))
|
||
|
||
for i in range(errored_cases):
|
||
result_repo.save_turn(Turn(
|
||
run_id=run.id,
|
||
case_id=f"err{i}",
|
||
round_index=1,
|
||
sent_message={"msgBody": {"content": "ping"}},
|
||
reply=None,
|
||
latency_ms=None,
|
||
))
|
||
|
||
all_total = total + connectivity_cases + errored_cases
|
||
all_passed = passed + connectivity_cases # 连通用例收到回复即通过（引擎口径）
|
||
run.summary = {
|
||
"total_cases": all_total,
|
||
"passed_cases": all_passed,
|
||
"failed_cases": all_total - all_passed,
|
||
"total_rules": total,
|
||
"passed_rules": passed,
|
||
"pass_rate": round(all_passed / all_total, 4) if all_total else 0.0,
|
||
}
|
||
RunRepository(session).update(run)
|
||
return run.id
|
||
|
||
|
||
# ── generate_report ───────────────────────────────────────────────────────
|
||
|
||
def test_generate_report_structure(report_session):
|
||
run_id = _seed_run(report_session)
|
||
report = generate_report(run_id, report_session)
|
||
|
||
assert report["run_id"] == run_id
|
||
assert report["target_name"] == "测试对象"
|
||
assert report["scenario_name"] == "测试场景"
|
||
assert report["status"] == "completed"
|
||
assert "summary" in report
|
||
assert "cases" in report
|
||
|
||
|
||
def test_generate_report_summary_values(report_session):
|
||
run_id = _seed_run(report_session, pass_rate=1.0, n_cases=2)
|
||
report = generate_report(run_id, report_session)
|
||
s = report["summary"]
|
||
assert s["total_cases"] == 2
|
||
assert s["passed_cases"] == 2
|
||
assert s["pass_rate"] == 1.0
|
||
|
||
|
||
def test_generate_report_partial_pass(report_session):
|
||
run_id = _seed_run(report_session, pass_rate=0.5, n_cases=2)
|
||
report = generate_report(run_id, report_session)
|
||
s = report["summary"]
|
||
assert s["passed_cases"] == 1
|
||
assert s["failed_cases"] == 1
|
||
|
||
|
||
def test_generate_report_cases_contain_turns_and_results(report_session):
|
||
run_id = _seed_run(report_session, n_cases=1)
|
||
report = generate_report(run_id, report_session)
|
||
assert len(report["cases"]) == 1
|
||
case = report["cases"][0]
|
||
assert len(case["turns"]) == 1
|
||
assert len(case["results"]) == 1
|
||
assert case["turns"][0]["sent_text"] == "问题0"
|
||
assert case["turns"][0]["latency_ms"] == 200
|
||
|
||
|
||
def test_generate_report_not_found_raises(report_session):
|
||
with pytest.raises(ValueError, match="run not found"):
|
||
generate_report("no-such-id", report_session)
|
||
|
||
|
||
# ── generate_compare_report ───────────────────────────────────────────────
|
||
|
||
def _scenario_of(session: Session, run_id: str) -> str:
|
||
return RunRepository(session).get(run_id).scenario_id
|
||
|
||
|
||
def test_compare_report_structure(report_session):
|
||
run_id_a = _seed_run(report_session, pass_rate=1.0, n_cases=2)
|
||
sid = _scenario_of(report_session, run_id_a)
|
||
run_id_b = _seed_run(report_session, pass_rate=0.5, n_cases=2, scenario_id=sid)
|
||
result = generate_compare_report(run_id_a, run_id_b, report_session)
|
||
|
||
assert "run_a" in result
|
||
assert "run_b" in result
|
||
assert "delta" in result
|
||
assert "cases" in result
|
||
assert result["run_a"]["run_id"] == run_id_a
|
||
assert result["run_b"]["run_id"] == run_id_b
|
||
|
||
|
||
def test_compare_report_delta(report_session):
|
||
run_id_a = _seed_run(report_session, pass_rate=0.5, n_cases=2)
|
||
sid = _scenario_of(report_session, run_id_a)
|
||
run_id_b = _seed_run(report_session, pass_rate=1.0, n_cases=2, scenario_id=sid)
|
||
result = generate_compare_report(run_id_a, run_id_b, report_session)
|
||
assert result["delta"]["pass_rate"] > 0 # B improved over A
|
||
|
||
|
||
def test_compare_report_changed_cases(report_session):
|
||
run_id_a = _seed_run(report_session, pass_rate=1.0, n_cases=2)
|
||
sid = _scenario_of(report_session, run_id_a)
|
||
run_id_b = _seed_run(report_session, pass_rate=0.5, n_cases=2, scenario_id=sid)
|
||
result = generate_compare_report(run_id_a, run_id_b, report_session)
|
||
# At least one case changed (A all-pass vs B half-pass)
|
||
assert result["changed_cases"] >= 1
|
||
|
||
|
||
def test_compare_report_case_level(report_session):
|
||
run_id_a = _seed_run(report_session, n_cases=1)
|
||
sid = _scenario_of(report_session, run_id_a)
|
||
run_id_b = _seed_run(report_session, n_cases=1, scenario_id=sid)
|
||
result = generate_compare_report(run_id_a, run_id_b, report_session)
|
||
assert len(result["cases"]) >= 1
|
||
case = result["cases"][0]
|
||
assert "run_a_passed" in case
|
||
assert "run_b_passed" in case
|
||
assert "changed" in case
|
||
|
||
|
||
def test_compare_report_different_scenarios_rejected(report_session):
|
||
run_id_a = _seed_run(report_session, n_cases=1)
|
||
run_id_b = _seed_run(report_session, n_cases=1) # separate scenario
|
||
with pytest.raises(ValueError, match="same scenario"):
|
||
generate_compare_report(run_id_a, run_id_b, report_session)
|
||
|
||
|
||
# ── connectivity case annotation (ticket 02) ─────────────────────────────
|
||
|
||
def test_report_marks_connectivity_case(report_session):
|
||
run_id = _seed_run(report_session, pass_rate=0.5, n_cases=2, connectivity_cases=1)
|
||
report = generate_report(run_id, report_session)
|
||
|
||
by_id = {c["case_id"]: c for c in report["cases"]}
|
||
assert by_id["conn0"]["connectivity"] is True
|
||
assert by_id["c0"]["connectivity"] is False
|
||
assert by_id["c1"]["connectivity"] is False
|
||
|
||
s = report["summary"]
|
||
assert s["connectivity_cases"] == 1
|
||
# 判定型通过率 = 判定型通过 1 ÷ 判定型总数 2
|
||
assert s["judged_pass_rate"] == 0.5
|
||
# 总通过率口径不变（含连通用例，ADR-0002）
|
||
assert s["pass_rate"] == round(2 / 3, 4)
|
||
|
||
|
||
def test_errored_case_not_marked_connectivity(report_session):
|
||
"""无判定结果但缺回复的用例是执行失败，不是连通用例。"""
|
||
run_id = _seed_run(report_session, n_cases=1, errored_cases=1)
|
||
report = generate_report(run_id, report_session)
|
||
|
||
by_id = {c["case_id"]: c for c in report["cases"]}
|
||
assert by_id["err0"]["connectivity"] is False
|
||
assert report["summary"]["connectivity_cases"] == 0
|
||
|
||
|
||
def test_all_connectivity_judged_pass_rate_none(report_session):
|
||
"""全为连通用例时判定型通过率为 None，不除零。"""
|
||
run_id = _seed_run(report_session, n_cases=0, connectivity_cases=2)
|
||
report = generate_report(run_id, report_session)
|
||
|
||
s = report["summary"]
|
||
assert s["connectivity_cases"] == 2
|
||
assert s["judged_pass_rate"] is None
|
||
|
||
|
||
def test_compare_report_marks_connectivity(report_session):
|
||
run_id_a = _seed_run(report_session, n_cases=1, connectivity_cases=1)
|
||
sid = _scenario_of(report_session, run_id_a)
|
||
run_id_b = _seed_run(report_session, n_cases=1, connectivity_cases=1, scenario_id=sid)
|
||
result = generate_compare_report(run_id_a, run_id_b, report_session)
|
||
|
||
by_id = {c["case_id"]: c for c in result["cases"]}
|
||
assert by_id["conn0"]["connectivity"] is True
|
||
assert by_id["c0"]["connectivity"] is False
|
||
# 连通用例双方均按引擎口径视为通过，不应标记 changed
|
||
assert by_id["conn0"]["run_a_passed"] is True
|
||
assert by_id["conn0"]["run_b_passed"] is True
|
||
assert by_id["conn0"]["changed"] is False
|
||
|
||
|
||
def test_markdown_report_shows_connectivity(report_session):
|
||
run_id = _seed_run(report_session, n_cases=1, connectivity_cases=1)
|
||
md = render_markdown_report(run_id, report_session)
|
||
assert "连通用例" in md
|
||
assert "判定型通过率" in md
|
||
|
||
|
||
# ── render_markdown_report ────────────────────────────────────────────────
|
||
|
||
def test_render_markdown_contains_header(report_session):
|
||
run_id = _seed_run(report_session)
|
||
md = render_markdown_report(run_id, report_session)
|
||
assert "# 评测报告" in md
|
||
|
||
|
||
def test_render_markdown_contains_summary_table(report_session):
|
||
run_id = _seed_run(report_session)
|
||
md = render_markdown_report(run_id, report_session)
|
||
assert "## 汇总" in md
|
||
assert "| 指标 | 数值 |" in md
|
||
assert "通过率" in md
|
||
|
||
|
||
def test_render_markdown_contains_case_section(report_session):
|
||
run_id = _seed_run(report_session, n_cases=1)
|
||
md = render_markdown_report(run_id, report_session)
|
||
assert "## 用例明细" in md
|
||
assert "### " in md # case header
|
||
|
||
|
||
def test_render_markdown_contains_rule_table(report_session):
|
||
run_id = _seed_run(report_session)
|
||
md = render_markdown_report(run_id, report_session)
|
||
assert "**规则评估结果**" in md
|
||
assert "keyword_match" in md
|
||
|
||
|
||
# ── render_json_report ────────────────────────────────────────────────────
|
||
|
||
def test_render_json_report_is_valid_json(report_session):
|
||
import json
|
||
run_id = _seed_run(report_session)
|
||
json_text = render_json_report(run_id, report_session)
|
||
parsed = json.loads(json_text)
|
||
assert parsed["run_id"] == run_id
|