0d074e5465
All checks were successful
CI / test (push) Successful in 4m3s
case 并行（默认 3）后原 5 case 场景在取消请求到达前就跑完， 取消语义不再被观测到。扩到 30 case（10 波）保证取消窗口充分。
335 lines
12 KiB
Python
335 lines
12 KiB
Python
"""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
|
||
|
||
import pytest
|
||
from agenteval.models import (
|
||
Case,
|
||
CaseType,
|
||
ChannelType,
|
||
EvalTarget,
|
||
PlatformType,
|
||
Scenario,
|
||
TargetStatus,
|
||
)
|
||
from agenteval.storage.repository import ScenarioRepository, TargetRepository
|
||
from agenteval.web.app import app
|
||
from httpx import ASGITransport, AsyncClient
|
||
|
||
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 (services.runs.execute_run) 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.evaluation import engine as engine_module
|
||
from agenteval.services import runs as run_service_module
|
||
from agenteval.storage import repository as repo_module
|
||
|
||
monkeypatch.setattr(db_module, "get_session", _test_get_session)
|
||
monkeypatch.setattr(run_service_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.
|
||
# Cases run 3-concurrent, so 30 cases ≈ 10 waves × reply_delay — comfortably
|
||
# longer than the pre-cancel window below.
|
||
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(30)
|
||
],
|
||
)
|
||
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 30 cases should have run.
|
||
assert mock_channel.send_calls < 30
|
||
|
||
|
||
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 set(logs.keys()) == {"turns", "results", "case_verdicts", "scenario_snapshot"}
|
||
|
||
# Characterization: exact turn shape, round-tripping the sent message.
|
||
assert len(logs["turns"]) == 1
|
||
turn = logs["turns"][0]
|
||
assert set(turn.keys()) == {
|
||
"id", "case_id", "round_index", "latency_ms",
|
||
"sent_text", "reply_text", "sent_at", "received_at",
|
||
}
|
||
assert turn["case_id"] == "c1"
|
||
assert turn["round_index"] == 1
|
||
assert turn["sent_text"] == "hi"
|
||
assert turn["reply_text"] # MockChannel always replies
|
||
assert turn["latency_ms"] is not None
|
||
assert turn["sent_at"] and turn["received_at"]
|
||
|
||
# case_verdicts are keyed by case id with passed/connectivity flags.
|
||
assert set(logs["case_verdicts"].keys()) == {"c1"}
|
||
assert set(logs["case_verdicts"]["c1"].keys()) == {"passed", "connectivity"}
|
||
|
||
# scenario_snapshot mirrors the seeded case's structure.
|
||
snap = logs["scenario_snapshot"]["c1"]
|
||
assert snap["id"] == "c1"
|
||
assert snap["type"] == "single"
|
||
assert snap["messages"] == ["hi"]
|
||
assert set(snap["expectations"].keys()) == {
|
||
"intent", "keywords_include", "keywords_exclude",
|
||
"response_time_max_ms", "coherence_min_score",
|
||
}
|
||
assert isinstance(snap["eval_rules"], list)
|
||
|
||
# results entries, when present, carry the rule outcome shape.
|
||
for entry in logs["results"]:
|
||
assert set(entry.keys()) == {"case_id", "rule_type", "passed", "score", "reason"}
|
||
|
||
|
||
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
|
||
|
||
|
||
async def test_cancel_run_without_live_task_marks_db_failed(client, seeded_db):
|
||
"""No live task (e.g. process restarted): cancel falls back to the DB row."""
|
||
from agenteval.models import EvalRun
|
||
from agenteval.storage.repository import RunRepository
|
||
|
||
run = RunRepository(seeded_db).create(EvalRun(target_id="t-1", scenario_id="s-1"))
|
||
|
||
resp = await client.post(f"/api/runs/{run.id}/cancel")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["status"] == "failed"
|
||
assert body["summary"]["error"]["code"] == "cancelled_by_user"
|
||
|
||
# Persisted, not just echoed.
|
||
got = (await client.get(f"/api/runs/{run.id}")).json()
|
||
assert got["status"] == "failed"
|
||
|
||
|
||
async def test_cancel_finished_run_rejected(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
|
||
|
||
resp = await client.post(f"/api/runs/{run_id}/cancel")
|
||
assert resp.status_code == 400
|
||
|
||
|
||
# ── triggered_by ─────────────────────────────────────────────────────────
|
||
|
||
async def test_start_run_default_triggered_by_manual(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
|
||
assert resp.json()["triggered_by"] == "manual"
|
||
|
||
|
||
async def test_start_run_ai_assistant_triggered_by(client, seeded_db, mock_channel):
|
||
resp = await client.post("/api/runs", json={
|
||
"target_id": "t-1", "scenario_id": "s-1", "triggered_by": "ai_assistant",
|
||
})
|
||
assert resp.status_code == 200
|
||
run_id = resp.json()["id"]
|
||
assert resp.json()["triggered_by"] == "ai_assistant"
|
||
|
||
# Persisted, not just echoed.
|
||
got = (await client.get(f"/api/runs/{run_id}")).json()
|
||
assert got["triggered_by"] == "ai_assistant"
|
||
|
||
|
||
async def test_start_run_invalid_triggered_by_422(client, seeded_db):
|
||
resp = await client.post("/api/runs", json={
|
||
"target_id": "t-1", "scenario_id": "s-1", "triggered_by": "robot",
|
||
})
|
||
assert resp.status_code == 422
|
||
|
||
|
||
async def test_list_runs_includes_names_and_trigger(client, seeded_db, mock_channel):
|
||
await client.post("/api/runs", json={
|
||
"target_id": "t-1", "scenario_id": "s-1", "triggered_by": "ai_assistant",
|
||
})
|
||
listing = (await client.get("/api/runs")).json()
|
||
assert len(listing) == 1
|
||
row = listing[0]
|
||
assert row["scenario_name"] == "mock-scenario"
|
||
assert row["target_name"] == "mock-target"
|
||
assert row["triggered_by"] == "ai_assistant"
|
||
|
||
|
||
# ── scenario_version snapshot (ticket 04) ────────────────────────────────
|
||
|
||
async def test_start_run_snapshots_scenario_version(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
|
||
assert resp.json()["scenario_version"] == 1
|
||
|
||
# 编辑考纲 → 场景升版 → 新运行快照新版本；旧运行保持 1
|
||
old_run_id = resp.json()["id"]
|
||
scenario = ScenarioRepository(seeded_db).get("s-1")
|
||
scenario.cases.append(Case(id="c2", type=CaseType.SINGLE, messages=["more"]))
|
||
ScenarioRepository(seeded_db).update(scenario)
|
||
|
||
resp2 = await client.post("/api/runs", json={
|
||
"target_id": "t-1", "scenario_id": "s-1", "triggered_by": "ai_assistant",
|
||
})
|
||
assert resp2.json()["scenario_version"] == 2
|
||
|
||
old = (await client.get(f"/api/runs/{old_run_id}")).json()
|
||
assert old["scenario_version"] == 1
|
||
|
||
listing = (await client.get("/api/runs")).json()
|
||
assert {row["scenario_version"] for row in listing} == {1, 2}
|