AgentEvalTool/tests/integration/test_reports_api.py

"""Integration tests for the reports API: /api/reports/{id}, /markdown, /compare."""

import json
import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine

from agenteval.models import (
    Case, CaseType, EvalResult, EvalRun, EvalTarget, RunStatus, Scenario,
    PlatformType, ChannelType, TargetStatus, Turn,
)
from agenteval.storage.repository import ResultRepository, RunRepository, ScenarioRepository, TargetRepository
from agenteval.web.app import app
from agenteval.web.deps import get_db


@pytest.fixture()
def client_with_db(tmp_path):
    from agenteval.storage.db import (  # noqa: F401
        EvalResultDB, EvalRunDB, EvalTargetDB, FileCategoryDB, FileRecordDB, ScenarioDB, TurnDB,
    )
    engine = create_engine(
        f"sqlite:///{tmp_path / 'reports_api.db'}",
        connect_args={"check_same_thread": False},
    )
    SQLModel.metadata.create_all(engine)
    session = Session(engine)

    def override_get_db():
        try:
            yield session
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    client = TestClient(app)
    yield client, session
    app.dependency_overrides.clear()
    session.close()


def _seed_run(session: Session, name_suffix: str = "", scenario_id: str | None = None) -> str:
    target = EvalTarget(
        name=f"target{name_suffix}",
        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=f"scenario{name_suffix}",
            cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
        )
        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)

    result_repo = ResultRepository(session)
    run_repo = RunRepository(session)

    turn = Turn(
        run_id=run.id, case_id="c1", round_index=1,
        sent_message={"msgBody": {"content": "问题"}},
        reply={"msgBody": {"content": "回答"}},
        latency_ms=300,
    )
    result_repo.save_turn(turn)
    db_turn = run_repo.get_turns(run.id)[0]

    result_repo.save_result(EvalResult(
        run_id=run.id, case_id="c1",
        turn_id=db_turn.id or "",
        rule_type="response_time",
        passed=True, score=0.9, reason="快",
    ))

    run.summary = {
        "total_cases": 1, "passed_cases": 1, "failed_cases": 0,
        "total_rules": 1, "passed_rules": 1, "pass_rate": 1.0,
    }
    RunRepository(session).update(run)
    return run.id


def test_get_report_200(client_with_db):
    client, session = client_with_db
    run_id = _seed_run(session)
    resp = client.get(f"/api/reports/{run_id}")
    assert resp.status_code == 200
    data = resp.json()
    assert data["run_id"] == run_id
    assert data["summary"]["pass_rate"] == 1.0


def test_get_report_404(client_with_db):
    client, _ = client_with_db
    resp = client.get("/api/reports/no-such-run")
    assert resp.status_code == 404


def test_get_html_report(client_with_db):
    client, session = client_with_db
    run_id = _seed_run(session)
    resp = client.get(f"/api/reports/{run_id}/html")
    assert resp.status_code == 200
    assert "text/html" in resp.headers["content-type"]
    assert "评测报告" in resp.text


def test_get_json_report(client_with_db):
    client, session = client_with_db
    run_id = _seed_run(session)
    resp = client.get(f"/api/reports/{run_id}/json")
    assert resp.status_code == 200
    data = json.loads(resp.text)
    assert "run_id" in data


def test_get_markdown_report(client_with_db):
    client, session = client_with_db
    run_id = _seed_run(session)
    resp = client.get(f"/api/reports/{run_id}/markdown")
    assert resp.status_code == 200
    assert "text/markdown" in resp.headers["content-type"]
    assert "# 评测报告" in resp.text
    assert "## 汇总" in resp.text


def test_get_markdown_report_attachment_header(client_with_db):
    client, session = client_with_db
    run_id = _seed_run(session)
    resp = client.get(f"/api/reports/{run_id}/markdown")
    assert "attachment" in resp.headers.get("content-disposition", "")


def test_compare_report(client_with_db):
    client, session = client_with_db
    run_id_a = _seed_run(session, "A")
    sid = RunRepository(session).get(run_id_a).scenario_id
    run_id_b = _seed_run(session, "B", scenario_id=sid)
    resp = client.get(f"/api/reports/compare?run1={run_id_a}&run2={run_id_b}")
    assert resp.status_code == 200
    data = resp.json()
    assert data["run_a"]["run_id"] == run_id_a
    assert data["run_b"]["run_id"] == run_id_b
    assert "delta" in data
    assert "cases" in data


def test_compare_report_different_scenarios_400(client_with_db):
    client, session = client_with_db
    run_id_a = _seed_run(session, "A")
    run_id_b = _seed_run(session, "B")  # separate scenario
    resp = client.get(f"/api/reports/compare?run1={run_id_a}&run2={run_id_b}")
    assert resp.status_code == 400
    assert "相同场景" in resp.json()["detail"]


def test_compare_report_run_not_found(client_with_db):
    client, session = client_with_db
    run_id = _seed_run(session)
    resp = client.get(f"/api/reports/compare?run1={run_id}&run2=ghost-id")
    assert resp.status_code == 404