69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
"""Gateway contract for the dedicated task deep-link history endpoint."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.gateway.routers.task_deeplink_threads import list_task_deeplink_threads
|
|
|
|
|
|
class _HistoryStore:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, str, int]] = []
|
|
|
|
async def list_threads(self, task_id: str, go_path: str, *, limit: int):
|
|
self.calls.append((task_id, go_path, limit))
|
|
return [
|
|
{
|
|
"thread_id": "thread-1",
|
|
"status": "idle",
|
|
"created_at": "2026-10-19T08:00:00+00:00",
|
|
"updated_at": "2026-10-19T08:01:00+00:00",
|
|
"metadata": {
|
|
"taskId": "108",
|
|
"taskRecordId": "108",
|
|
"taskGoPath": "rwfx",
|
|
},
|
|
"values": {"title": "任务分析"},
|
|
"interrupts": {},
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_endpoint_trims_task_id_and_preserves_thread_shape():
|
|
store = _HistoryStore()
|
|
request = SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(task_deeplink_thread_store=store),
|
|
)
|
|
)
|
|
|
|
response = await list_task_deeplink_threads(
|
|
request,
|
|
task_id=" 108 ",
|
|
go_path="rwfx",
|
|
limit=25,
|
|
)
|
|
|
|
assert store.calls == [("108", "rwfx", 25)]
|
|
assert response.model_dump() == {
|
|
"threads": [
|
|
{
|
|
"thread_id": "thread-1",
|
|
"status": "idle",
|
|
"created_at": "2026-10-19T08:00:00+00:00",
|
|
"updated_at": "2026-10-19T08:01:00+00:00",
|
|
"metadata": {
|
|
"taskId": "108",
|
|
"taskRecordId": "108",
|
|
"taskGoPath": "rwfx",
|
|
},
|
|
"values": {"title": "任务分析"},
|
|
"interrupts": {},
|
|
}
|
|
]
|
|
}
|