AgentEvalTool/tests/unit/test_report.py

"""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,
) -> str:
    """Create a minimal completed run with real data in the DB and return run_id."""
    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)],
        )
        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)

    run.summary = {
        "total_cases": total,
        "passed_cases": passed,
        "failed_cases": total - passed,
        "total_rules": total,
        "passed_rules": passed,
        "pass_rate": round(pass_rate, 4),
    }
    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)


# ── 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