AgentEvalTool/tests/integration/test_runs_api.py

"""Integration tests for the /api/runs endpoints.

Uses ``httpx.AsyncClient`` with ``app=`` to drive the FastAPI app in-process
(no real server). The engine's channel is monkeypatched to a MockChannel so
no network calls are made.
"""

import asyncio
from typing import Any

import pytest
from httpx import ASGITransport, AsyncClient

from agenteval.channels.base import SendResult
from agenteval.models import (
    Case, CaseType, ChannelType, EvalTarget, PlatformType,
    Scenario, TargetStatus,
)
from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
from agenteval.web.app import app
from tests.unit.mock_channel import MockChannel


@pytest.fixture()
def seeded_db(db_session, monkeypatch):
    """Patch the global get_session to return the test session, then seed a
    target + scenario. Also skip init_db so the real data/ DB is not touched."""
    from agenteval.storage import db as db_module
    from agenteval.web import app as app_module

    # Skip init_db in lifespan (it would create tables in the real data/ DB).
    monkeypatch.setattr(app_module, "init_db", lambda: None)

    # Background task (_run_evaluation) calls get_session() directly.
    # Because `from agenteval.storage.db import get_session` binds a local
    # reference in every importing module, we must patch every consumer.
    def _test_get_session():
        return db_session

    from agenteval.storage import db as db_module
    from agenteval.web.routers import runs as runs_module
    from agenteval.storage import repository as repo_module
    from agenteval.evaluation import engine as engine_module

    monkeypatch.setattr(db_module, "get_session", _test_get_session)
    monkeypatch.setattr(runs_module, "get_session", _test_get_session)
    monkeypatch.setattr(repo_module, "get_session", _test_get_session)
    monkeypatch.setattr(engine_module, "get_session", _test_get_session)

    # FastAPI endpoints use Depends(get_db). Override the dependency.
    from agenteval.web.deps import get_db

    def _test_get_db():
        try:
            yield db_session
        finally:
            pass  # Don't close the fixture-owned session.

    app.dependency_overrides[get_db] = _test_get_db

    target = EvalTarget(
        id="t-1", name="mock-target",
        platform=PlatformType.AI_DIGITAL_EMPLOYEE,
        channel_type=ChannelType.TUTU_API,
        channel_config={
            "base_url": "http://mock", "token": "x",
            "tenant": "t", "chat_channel_id": "c", "chat_contact_id": "u",
        },
        status=TargetStatus.ACTIVE,
    )
    TargetRepository(db_session).create(target)

    scenario = Scenario(
        id="s-1", name="mock-scenario",
        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
    )
    ScenarioRepository(db_session).create(scenario)

    yield db_session

    app.dependency_overrides.clear()


@pytest.fixture()
def mock_channel(monkeypatch):
    """Make ChannelFactory.create return a MockChannel for every target."""
    channel = MockChannel(reply_delay=0.02)

    def _fake_create(target):
        return channel

    from agenteval.channels import factory as factory_module
    monkeypatch.setattr(factory_module.ChannelFactory, "create", _fake_create)
    return channel


@pytest.fixture()
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


# ── list / get ───────────────────────────────────────────────────────────

async def test_list_runs_empty(client, seeded_db):
    resp = await client.get("/api/runs")
    assert resp.status_code == 200
    assert resp.json() == []


async def test_start_run_then_get(client, seeded_db, mock_channel):
    resp = await client.post("/api/runs", json={
        "target_id": "t-1", "scenario_id": "s-1",
    })
    assert resp.status_code == 200
    body = resp.json()
    run_id = body["id"]
    assert body["status"] == "pending"

    # Wait for the background task to finish.
    for _ in range(50):
        await asyncio.sleep(0.05)
        r = await client.get(f"/api/runs/{run_id}")
        if r.json()["status"] in ("completed", "failed"):
            break

    final = (await client.get(f"/api/runs/{run_id}")).json()
    assert final["status"] == "completed"
    assert final["summary"]["total_cases"] == 1
    assert mock_channel.send_calls == 1


async def test_cancel_run(client, seeded_db, mock_channel):
    # Seed a multi-case scenario so the run takes long enough to cancel.
    from agenteval.models import Case, CaseType, Scenario
    from agenteval.storage.repository import ScenarioRepository
    multi = Scenario(
        id="s-long", name="long-scenario",
        cases=[
            Case(id=f"lc{i}", type=CaseType.SINGLE, messages=[f"m{i}"])
            for i in range(5)
        ],
    )
    ScenarioRepository(seeded_db).create(multi)

    # Use a slow channel so we have time to cancel mid-flight.
    mock_channel.reply_delay = 0.1

    resp = await client.post("/api/runs", json={
        "target_id": "t-1", "scenario_id": "s-long",
    })
    run_id = resp.json()["id"]

    # Give the task a moment to start and process at least one case.
    await asyncio.sleep(0.15)

    cancel = await client.post(f"/api/runs/{run_id}/cancel")
    assert cancel.status_code == 200

    # Wait for the task to observe the cancel.
    for _ in range(50):
        await asyncio.sleep(0.05)
        r = await client.get(f"/api/runs/{run_id}")
        if r.json()["status"] == "failed":
            break

    final = (await client.get(f"/api/runs/{run_id}")).json()
    assert final["status"] == "failed"
    assert final["summary"]["error"]["code"] == "cancelled_by_user"
    # Not all 5 cases should have run.
    assert mock_channel.send_calls < 5


async def test_get_run_logs(client, seeded_db, mock_channel):
    resp = await client.post("/api/runs", json={
        "target_id": "t-1", "scenario_id": "s-1",
    })
    run_id = resp.json()["id"]

    for _ in range(50):
        await asyncio.sleep(0.05)
        r = await client.get(f"/api/runs/{run_id}")
        if r.json()["status"] == "completed":
            break

    logs = (await client.get(f"/api/runs/{run_id}/logs")).json()
    assert "turns" in logs
    assert "results" in logs
    assert "scenario_snapshot" in logs
    assert len(logs["turns"]) == 1


async def test_start_run_missing_target(client, seeded_db):
    resp = await client.post("/api/runs", json={
        "target_id": "does-not-exist", "scenario_id": "s-1",
    })
    assert resp.status_code == 404