163 lines
6.9 KiB
Python
163 lines
6.9 KiB
Python
"""取消语义的 MySQL 集成测试（文档 §6.4）。
|
||
|
||
覆盖三条真实链路：
|
||
1. 取消 `queued` 运行 → `202`，`agent_run.status=cancel_requested`，
|
||
且原请求在 `request_idempotency` 落 `failed + RUN_CANCELLED`；
|
||
2. **重复取消**（换幂等键、真实 HTTP）→ 返回同一状态，**不报错**；
|
||
3. 已成功 / 已失败的运行取消 → `409 RUN_NOT_CANCELLABLE`。
|
||
|
||
依赖真实 MySQL；所有写入在 finally 中按 session_id 清理。
|
||
"""
|
||
|
||
from uuid import uuid4
|
||
|
||
import httpx
|
||
import pytest
|
||
from sqlalchemy import delete, select
|
||
|
||
from app.api.dependencies.auth import build_request_context
|
||
from app.core.contracts import AgentRequest, RequestContext
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.main import create_app
|
||
from app.model.audit import InteractionAudit
|
||
from app.model.conversation import ConversationMessage
|
||
from app.model.platform import AgentRun, DomainEventOutbox, OutboxDelivery, RequestIdempotency
|
||
from app.service.agent_run_application_service import AgentRunApplicationService
|
||
from app.worker.runtime import WorkerRuntime
|
||
|
||
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("acceptance_registry")]
|
||
|
||
|
||
async def state_of(run_id: str) -> tuple[str, str | None, str]:
|
||
async with SessionFactory() as session:
|
||
run = await session.scalar(select(AgentRun).where(AgentRun.run_id == run_id))
|
||
assert run is not None
|
||
idem = await session.get(RequestIdempotency, run.idempotency_id)
|
||
assert idem is not None
|
||
return run.status, idem.error_code, idem.status
|
||
|
||
|
||
async def cleanup(session_id: str, run_id: str) -> None:
|
||
async with SessionFactory() as session, session.begin():
|
||
event_ids = select(DomainEventOutbox.event_id).where(
|
||
DomainEventOutbox.aggregate_id == run_id)
|
||
await session.execute(delete(OutboxDelivery).where(OutboxDelivery.event_id.in_(event_ids)))
|
||
await session.execute(delete(DomainEventOutbox).where(
|
||
DomainEventOutbox.aggregate_id == run_id))
|
||
await session.execute(delete(InteractionAudit).where(
|
||
InteractionAudit.session_id == session_id))
|
||
await session.execute(delete(AgentRun).where(AgentRun.session_id == session_id))
|
||
await session.execute(delete(RequestIdempotency).where(
|
||
RequestIdempotency.session_id == session_id))
|
||
await session.execute(delete(ConversationMessage).where(
|
||
ConversationMessage.session_id == session_id))
|
||
|
||
|
||
def context() -> RequestContext:
|
||
return RequestContext(user_id="1", trace_id=str(uuid4()), roles=("customer",),
|
||
permissions=("agent:run", "agent:cancel"))
|
||
|
||
|
||
def client_for(ctx: RequestContext) -> httpx.AsyncClient:
|
||
application = create_app()
|
||
application.dependency_overrides[build_request_context] = lambda: ctx
|
||
return httpx.AsyncClient(transport=httpx.ASGITransport(app=application),
|
||
base_url="http://test")
|
||
|
||
|
||
async def submit(ctx: RequestContext, session_id: str) -> str:
|
||
async with SessionFactory() as session:
|
||
accepted = await AgentRunApplicationService(session).accept(
|
||
AgentRequest(agent_type="customer_service", message="cancel integration",
|
||
session_id=session_id, idempotency_key=str(uuid4())),
|
||
ctx,
|
||
)
|
||
return accepted.run_id
|
||
|
||
|
||
async def test_cancel_is_idempotent_and_terminates_original_request() -> None:
|
||
ctx = context()
|
||
session_id = f"cancel-{uuid4()}"
|
||
run_id = ""
|
||
try:
|
||
run_id = await submit(ctx, session_id)
|
||
async with client_for(ctx) as client:
|
||
first = await client.post(f"/api/v1/agent-runs/{run_id}/cancellations",
|
||
json={"reason": "user_cancelled"},
|
||
headers={"Idempotency-Key": str(uuid4())})
|
||
# 换一个幂等键重复取消：必须幂等返回同一状态，而不是 409。
|
||
second = await client.post(f"/api/v1/agent-runs/{run_id}/cancellations",
|
||
json={"reason": "user_cancelled"},
|
||
headers={"Idempotency-Key": str(uuid4())})
|
||
|
||
assert first.status_code == 202
|
||
assert second.status_code == 202
|
||
assert first.json()["data"] == second.json()["data"]
|
||
assert first.json()["data"]["status"] == "cancel_requested"
|
||
assert first.json()["data"]["run_id"] == run_id
|
||
|
||
status, error_code, idem_status = await state_of(run_id)
|
||
assert status == "cancel_requested"
|
||
# 文档 §6.4：原请求以 failed + RUN_CANCELLED 结束，不扩展状态枚举。
|
||
assert (idem_status, error_code) == ("failed", "RUN_CANCELLED")
|
||
finally:
|
||
if run_id:
|
||
await cleanup(session_id, run_id)
|
||
|
||
|
||
async def test_cancel_after_worker_completion_is_not_cancellable(
|
||
acceptance_registry,
|
||
) -> None:
|
||
ctx = context()
|
||
session_id = f"cancel-done-{uuid4()}"
|
||
run_id = ""
|
||
runtime = WorkerRuntime(
|
||
acceptance_registry, resolve_identity=lambda value: _identity(value, ctx)
|
||
)
|
||
try:
|
||
run_id = await submit(ctx, session_id)
|
||
await runtime.execute(run_id)
|
||
status, _error_code, _idem = await state_of(run_id)
|
||
assert status == "succeeded"
|
||
|
||
async with client_for(ctx) as client:
|
||
response = await client.post(f"/api/v1/agent-runs/{run_id}/cancellations",
|
||
json={"reason": "user_cancelled"},
|
||
headers={"Idempotency-Key": str(uuid4())})
|
||
|
||
assert response.status_code == 409
|
||
assert response.json()["error"]["code"] == "RUN_NOT_CANCELLABLE"
|
||
assert response.json()["error"]["retryable"] is False
|
||
finally:
|
||
if run_id:
|
||
await cleanup(session_id, run_id)
|
||
|
||
|
||
async def test_worker_advances_cancel_requested_to_cancelled_with_idempotency_marker() -> None:
|
||
"""worker 真正落 `cancelled` 时，幂等记录仍必须是 `failed + RUN_CANCELLED`。"""
|
||
ctx = context()
|
||
session_id = f"cancel-worker-{uuid4()}"
|
||
run_id = ""
|
||
try:
|
||
run_id = await submit(ctx, session_id)
|
||
async with client_for(ctx) as client:
|
||
response = await client.post(f"/api/v1/agent-runs/{run_id}/cancellations",
|
||
json={"reason": "user_cancelled"},
|
||
headers={"Idempotency-Key": str(uuid4())})
|
||
assert response.status_code == 202
|
||
|
||
runtime = WorkerRuntime(resolve_identity=lambda value: _identity(value, ctx))
|
||
await runtime.execute(run_id)
|
||
|
||
status, error_code, idem_status = await state_of(run_id)
|
||
assert status == "cancelled"
|
||
assert (idem_status, error_code) == ("failed", "RUN_CANCELLED")
|
||
finally:
|
||
if run_id:
|
||
await cleanup(session_id, run_id)
|
||
|
||
|
||
async def _identity(value: RequestContext, fallback: RequestContext) -> RequestContext:
|
||
del value
|
||
return fallback
|