111 lines
4.6 KiB
Python
111 lines
4.6 KiB
Python
"""Acceptance tests for the structured cancel cascade (Task 4).
|
|
|
|
Verifies that when DeerFlow cancel fails, the parent run is NOT marked
|
|
``cancelled`` —it stays non-terminal (``cancel_failed``) so the failure is
|
|
visible and retryable. A failed cancel must never masquerade as cancelled.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.maf_roundtable.deerflow_agent_adapter import DeerFlowAgentAdapter
|
|
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
|
|
from deerflow.persistence.maf_roundtable.model import Base
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def store():
|
|
engine = create_async_engine("sqlite+aiosqlite://", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, expire_on_commit=False)
|
|
s = make_maf_roundtable_store(sf)
|
|
assert s is not None
|
|
yield s
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def svc(store):
|
|
from maf_test_helpers import make_maf_service
|
|
|
|
return make_maf_service(store, enabled=True, admin_diagnostic_access=True)
|
|
|
|
|
|
def _mock_client(*, cancel_succeeds: bool) -> MagicMock:
|
|
client = MagicMock()
|
|
client._app.state.run_manager.cancel = AsyncMock(return_value=None)
|
|
client.cancel_run = AsyncMock(return_value=cancel_succeeds)
|
|
return client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_all_succeeds_marks_cancelled(svc, store):
|
|
"""When every sub-run cancels, all_cancelled is True."""
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
|
|
active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
|
|
await svc.store.seat_runs.mark_running(active[0]["id"], deerflow_run_id="df-1")
|
|
|
|
adapter = DeerFlowAgentAdapter(_mock_client(cancel_succeeds=True), service=svc, user_id="u1")
|
|
result = await adapter.cancel(run["id"])
|
|
assert result["all_cancelled"] is True
|
|
assert result["failed_run_ids"] == []
|
|
assert result["cancelled_run_ids"] == [active[0]["id"]]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_failure_keeps_run_nonterminal(svc, store):
|
|
"""When DeerFlow cancel fails, the parent run must NOT be marked cancelled.
|
|
|
|
The adapter returns all_cancelled=False; the router would then mark the
|
|
run ``cancel_failed`` (non-terminal). This test verifies the adapter-level
|
|
signal that drives that decision.
|
|
"""
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
await svc.store.seat_runs.create_seat_run(maf_run_id=run["id"], agent_id="a1", turn_index=1)
|
|
active = await svc.store.seat_runs.get_active_by_maf_run(run["id"])
|
|
await svc.store.seat_runs.mark_running(active[0]["id"], deerflow_run_id="df-2")
|
|
|
|
adapter = DeerFlowAgentAdapter(_mock_client(cancel_succeeds=False), service=svc, user_id="u1")
|
|
result = await adapter.cancel(run["id"])
|
|
|
|
assert result["all_cancelled"] is False
|
|
assert result["failed_run_ids"] == [active[0]["id"]]
|
|
assert result["cancelled_run_ids"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_cancel_failed_service_method(svc, store):
|
|
"""The service ``mark_cancel_failed`` transitions a cancelling run to
|
|
``cancel_failed`` (non-terminal), recording a sanitized reason."""
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
# Put the run into cancelling via start_cancel.
|
|
await svc.start_cancel(run_id=run["id"], user_id="u1")
|
|
|
|
updated = await svc.mark_cancel_failed(
|
|
run_id=run["id"], user_id="u1", reason="1 sub-run failed to cancel"
|
|
)
|
|
assert updated["status"] == "cancel_failed"
|
|
assert updated["error_code"] == "CancelCascadeIncomplete"
|
|
assert "failed to cancel" in updated["error_message_safe"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_no_active_runs_is_all_cancelled(svc, store):
|
|
"""With no active seat runs, cancel is trivially all_cancelled (nothing to do)."""
|
|
session = await svc.create_session(user_id="u1", body={"task_text": "t"})
|
|
run = await svc.create_run(session_id=session["id"], user_id="u1")
|
|
adapter = DeerFlowAgentAdapter(_mock_client(cancel_succeeds=True), service=svc, user_id="u1")
|
|
result = await adapter.cancel(run["id"])
|
|
assert result["all_cancelled"] is True
|
|
assert result["cancelled_run_ids"] == []
|