195 lines
7.3 KiB
Python
195 lines
7.3 KiB
Python
"""Integration tests for exploration settlement on campaign finalize (v0.9 票据 07).
|
|
|
|
Covers both terminal transitions: window completion (durable loop COMPLETE)
|
|
and mid-flight cancellation. Running sessions must turn expired and reject
|
|
further messages (409), leaving no dangling sessions.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from agenteval.evaluation import campaign_runner as runner_module
|
|
from agenteval.evaluation.campaign_runner import CampaignRuntime
|
|
from agenteval.exploration.models import ExplorationSessionStatus
|
|
from agenteval.models import (
|
|
Campaign,
|
|
CampaignPlanEntry,
|
|
Case,
|
|
CaseType,
|
|
ChannelType,
|
|
EvalTarget,
|
|
PlatformType,
|
|
Scenario,
|
|
TargetStatus,
|
|
)
|
|
from agenteval.storage.db import utc_now
|
|
from agenteval.storage.repository import (
|
|
CampaignRepository,
|
|
ExplorationSessionRepository,
|
|
ScenarioRepository,
|
|
TargetRepository,
|
|
)
|
|
from agenteval.web.app import app
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
|
|
def _session_payload(campaign_id: str) -> dict:
|
|
return {
|
|
"campaign_id": campaign_id,
|
|
"persona": {"name": "巡检用户", "traits": ["耐心"]},
|
|
"goal": "查询账单",
|
|
"triggered_by": "manual",
|
|
}
|
|
|
|
|
|
@pytest.fixture()
|
|
def seeded_db(db_session, monkeypatch):
|
|
"""Point every get_session consumer at the test session and seed the
|
|
target + scenario needed by both the loop path and the cancel path."""
|
|
from agenteval.channels import factory as factory_module
|
|
from agenteval.evaluation import engine as engine_module
|
|
from agenteval.exploration import lifecycle as exploration_lifecycle
|
|
from agenteval.storage import db as db_module
|
|
from agenteval.storage import repository as repo_module
|
|
from agenteval.web import app as app_module
|
|
|
|
monkeypatch.setattr(app_module, "init_db", lambda: None)
|
|
|
|
def _test_get_session():
|
|
return db_session
|
|
|
|
monkeypatch.setattr(db_module, "get_session", _test_get_session)
|
|
monkeypatch.setattr(repo_module, "get_session", _test_get_session)
|
|
monkeypatch.setattr(engine_module, "get_session", _test_get_session)
|
|
monkeypatch.setattr(runner_module, "get_session", _test_get_session)
|
|
monkeypatch.setattr(runner_module, "_auto_start_analysis", lambda campaign, session: None)
|
|
|
|
from tests.unit.mock_channel import MockChannel
|
|
|
|
channel = MockChannel(reply_delay=0.0)
|
|
monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel)
|
|
monkeypatch.setattr(exploration_lifecycle, "start_judge_review", lambda session_id: None)
|
|
|
|
from agenteval.web.deps import get_db
|
|
|
|
def _test_get_db():
|
|
try:
|
|
yield db_session
|
|
finally:
|
|
pass
|
|
|
|
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"},
|
|
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()
|
|
|
|
|
|
def _make_campaign(db_session, campaign_id: str, *, time_scale: float) -> Campaign:
|
|
return CampaignRepository(db_session).create(
|
|
Campaign(
|
|
id=campaign_id,
|
|
name=f"campaign-{campaign_id}",
|
|
target_id="t-1",
|
|
window_seconds=3600,
|
|
time_scale=time_scale,
|
|
plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)],
|
|
status="running",
|
|
started_at=utc_now(),
|
|
)
|
|
)
|
|
|
|
|
|
async def _create_running_session(db_session, campaign_id: str) -> str:
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
resp = await c.post("/api/exploration/sessions", json=_session_payload(campaign_id))
|
|
assert resp.status_code == 200, resp.text
|
|
session_id = resp.json()["id"]
|
|
assert ExplorationSessionRepository(db_session).get(session_id).status == (ExplorationSessionStatus.RUNNING)
|
|
return session_id
|
|
|
|
|
|
async def test_window_completion_expires_running_sessions(seeded_db):
|
|
campaign = _make_campaign(seeded_db, "c-done", time_scale=3600.0)
|
|
session_id = await _create_running_session(seeded_db, "c-done")
|
|
|
|
# time_scale 3600 → 1 real second covers the whole window.
|
|
runtime = CampaignRuntime(session_factory=lambda: seeded_db, tick_seconds=0.05)
|
|
try:
|
|
assert runtime.start(campaign.id)
|
|
async with asyncio.timeout(5):
|
|
while CampaignRepository(seeded_db).get(campaign.id).status.value == "running":
|
|
await asyncio.sleep(0.05)
|
|
finally:
|
|
await runtime.shutdown()
|
|
|
|
assert CampaignRepository(seeded_db).get("c-done").status.value == "completed"
|
|
settled = ExplorationSessionRepository(seeded_db).get(session_id)
|
|
assert settled.status == ExplorationSessionStatus.EXPIRED
|
|
assert settled.closed_at is not None
|
|
|
|
|
|
async def test_cancel_expires_running_sessions_and_rejects_messages(seeded_db):
|
|
_make_campaign(seeded_db, "c-1", time_scale=1.0)
|
|
session_id = await _create_running_session(seeded_db, "c-1")
|
|
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
resp = await client.post("/api/campaigns/c-1/cancel")
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["status"] == "cancelled"
|
|
|
|
settled = ExplorationSessionRepository(seeded_db).get(session_id)
|
|
assert settled.status == ExplorationSessionStatus.EXPIRED
|
|
assert settled.closed_at is not None
|
|
|
|
resp = await client.post(f"/api/exploration/sessions/{session_id}/messages", json={"content": "还在吗"})
|
|
assert resp.status_code == 409
|
|
|
|
|
|
async def test_completed_session_survives_settlement(seeded_db):
|
|
_make_campaign(seeded_db, "c-1", time_scale=1.0)
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
resp = await client.post("/api/exploration/sessions", json=_session_payload("c-1"))
|
|
session_id = resp.json()["id"]
|
|
resp = await client.post(
|
|
f"/api/exploration/sessions/{session_id}/close",
|
|
json={"experience": {"goal_achieved": True}},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
await client.post("/api/campaigns/c-1/cancel")
|
|
|
|
assert ExplorationSessionRepository(seeded_db).get(session_id).status == (ExplorationSessionStatus.COMPLETED)
|
|
|
|
|
|
async def test_production_line_manual_trigger_also_settles(seeded_db):
|
|
_make_campaign(seeded_db, "c-1", time_scale=1.0)
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
resp = await client.post(
|
|
"/api/exploration/sessions",
|
|
json={**_session_payload("c-1"), "triggered_by": "manual"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
session_id = resp.json()["id"]
|
|
assert resp.json()["triggered_by"] == "manual"
|
|
|
|
# 正式线手动会话同样纳入终态结算，触发来源不影响收口。
|
|
await client.post("/api/campaigns/c-1/cancel")
|
|
|
|
assert ExplorationSessionRepository(seeded_db).get(session_id).status == (ExplorationSessionStatus.EXPIRED)
|