123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
import asyncio
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway.routers import parallel_agent_tasks
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.parallel_agent_tasks import ParallelAgentTaskRepository
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parallel_task_repository_rejects_stale_revision(tmp_path) -> None:
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'strategy-project-cas.db'}")
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(Base.metadata.create_all)
|
|
store = ParallelAgentTaskRepository(async_sessionmaker(engine, expire_on_commit=False))
|
|
try:
|
|
original = await store.create_task(
|
|
"user-1",
|
|
{
|
|
"id": "strategy-selection-cas",
|
|
"title": "original",
|
|
"messages": [{"snapshot": "original"}],
|
|
},
|
|
)
|
|
await asyncio.sleep(0.002)
|
|
newer = await store.update_task(
|
|
"strategy-selection-cas",
|
|
"user-1",
|
|
title="newer",
|
|
messages=[{"snapshot": "newer"}],
|
|
expected_updated_at=original["updated_at"],
|
|
)
|
|
assert newer is not None
|
|
|
|
stale = await store.update_task(
|
|
"strategy-selection-cas",
|
|
"user-1",
|
|
title="stale",
|
|
messages=[{"snapshot": "stale"}],
|
|
expected_updated_at=original["updated_at"],
|
|
)
|
|
assert stale is None
|
|
final = await store.get_task("strategy-selection-cas", "user-1")
|
|
assert final is not None
|
|
assert final["title"] == "newer"
|
|
assert final["messages"] == [{"snapshot": "newer"}]
|
|
|
|
lightweight = await store.list_tasks("user-1")
|
|
assert "messages" not in lightweight[0]
|
|
with_snapshots = await store.list_tasks("user-1", include_messages=True)
|
|
assert with_snapshots[0]["messages"] == [{"snapshot": "newer"}]
|
|
|
|
await store.create_task(
|
|
"user-1",
|
|
{
|
|
"id": "parallel-unrelated",
|
|
"title": "unrelated",
|
|
"messages": [{"large": "transcript"}],
|
|
},
|
|
)
|
|
await asyncio.sleep(0.002)
|
|
await store.create_task(
|
|
"user-1",
|
|
{
|
|
"id": "strategy-selection-second",
|
|
"title": "second",
|
|
"messages": [{"snapshot": "second"}],
|
|
},
|
|
)
|
|
first_page = await store.list_tasks(
|
|
"user-1",
|
|
limit=1,
|
|
id_prefix="strategy-selection-",
|
|
include_messages=True,
|
|
)
|
|
second_page = await store.list_tasks(
|
|
"user-1",
|
|
limit=1,
|
|
offset=1,
|
|
id_prefix="strategy-selection-",
|
|
include_messages=True,
|
|
)
|
|
assert [first_page[0]["id"], second_page[0]["id"]] == [
|
|
"strategy-selection-second",
|
|
"strategy-selection-cas",
|
|
]
|
|
assert first_page[0]["messages"] == [{"snapshot": "second"}]
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parallel_task_route_reports_revision_conflict() -> None:
|
|
class ConflictStore:
|
|
async def update_task(self, *_args, **_kwargs):
|
|
return None
|
|
|
|
async def get_task(self, *_args, **_kwargs):
|
|
return {"id": "strategy-selection-cas"}
|
|
|
|
request = SimpleNamespace(
|
|
state=SimpleNamespace(user=SimpleNamespace(id="user-1")),
|
|
app=SimpleNamespace(state=SimpleNamespace(parallel_agent_task_store=ConflictStore())),
|
|
)
|
|
body = parallel_agent_tasks.TaskUpdateRequest(
|
|
title="stale",
|
|
expected_updated_at=datetime.now(UTC),
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as error:
|
|
await parallel_agent_tasks.update_task(
|
|
request,
|
|
"strategy-selection-cas",
|
|
body,
|
|
)
|
|
|
|
assert error.value.status_code == 409
|
|
assert "其他页面更新" in str(error.value.detail)
|