187 lines
4.9 KiB
Python
187 lines
4.9 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
|
|
from deerflow.persistence.task_deeplink_threads.base import (
|
|
MAX_TASK_DEEPLINK_SUB_TYPE,
|
|
task_deeplink_binding,
|
|
)
|
|
|
|
|
|
class _HistoryStore:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, str, int, str | None]] = []
|
|
|
|
async def list_threads(
|
|
self,
|
|
task_id: str,
|
|
go_path: str,
|
|
*,
|
|
limit: int,
|
|
sub_type: str | None = None,
|
|
):
|
|
self.calls.append((task_id, go_path, limit, sub_type))
|
|
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": {},
|
|
}
|
|
]
|
|
|
|
|
|
def _request_with(store: _HistoryStore) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
app=SimpleNamespace(
|
|
state=SimpleNamespace(task_deeplink_thread_store=store),
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_endpoint_trims_task_id_and_preserves_thread_shape():
|
|
store = _HistoryStore()
|
|
|
|
response = await list_task_deeplink_threads(
|
|
_request_with(store),
|
|
task_id=" 108 ",
|
|
go_path="rwfx",
|
|
limit=25,
|
|
sub_type=None,
|
|
)
|
|
|
|
assert store.calls == [("108", "rwfx", 25, None)]
|
|
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": {},
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_endpoint_passes_sub_type_through():
|
|
store = _HistoryStore()
|
|
|
|
await list_task_deeplink_threads(
|
|
_request_with(store),
|
|
task_id="task-1",
|
|
go_path="b5-1",
|
|
limit=100,
|
|
sub_type=" yuqing ",
|
|
)
|
|
# Explicitly empty sub_type must survive as "" (the no-type partition),
|
|
# not collapse to None (which would mean "every sub-type").
|
|
await list_task_deeplink_threads(
|
|
_request_with(store),
|
|
task_id="task-1",
|
|
go_path="b5-1",
|
|
limit=100,
|
|
sub_type="",
|
|
)
|
|
|
|
assert store.calls == [
|
|
("task-1", "b5-1", 100, "yuqing"),
|
|
("task-1", "b5-1", 100, ""),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_endpoint_accepts_a_custom_mapping_go_path() -> None:
|
|
store = _HistoryStore()
|
|
|
|
await list_task_deeplink_threads(
|
|
_request_with(store),
|
|
task_id="task-1",
|
|
go_path="区域研判",
|
|
limit=100,
|
|
sub_type=None,
|
|
)
|
|
|
|
assert store.calls == [("task-1", "区域研判", 100, None)]
|
|
|
|
|
|
def test_binding_extracts_b51_task_type_as_sub_type():
|
|
binding = task_deeplink_binding(
|
|
{
|
|
"taskId": "task-1",
|
|
"taskRecordId": "task-1",
|
|
"taskGoPath": "b5-1",
|
|
"taskType": "yuqing",
|
|
}
|
|
)
|
|
|
|
assert binding is not None
|
|
assert binding["sub_type"] == "yuqing"
|
|
|
|
|
|
def test_binding_defaults_missing_sub_type_to_empty_string():
|
|
for go_path in ("3qfx", "rwfx", "xdfx", "b5-1"):
|
|
binding = task_deeplink_binding(
|
|
{
|
|
"taskId": "task-1",
|
|
"taskRecordId": "task-1",
|
|
"taskGoPath": go_path,
|
|
}
|
|
)
|
|
assert binding is not None
|
|
assert binding["sub_type"] == ""
|
|
|
|
|
|
def test_binding_caps_sub_type_to_column_width():
|
|
binding = task_deeplink_binding(
|
|
{
|
|
"taskId": "task-1",
|
|
"taskRecordId": "task-1",
|
|
"taskGoPath": "b5-1",
|
|
"taskType": "x" * (MAX_TASK_DEEPLINK_SUB_TYPE + 10),
|
|
}
|
|
)
|
|
|
|
assert binding is not None
|
|
assert len(binding["sub_type"]) == MAX_TASK_DEEPLINK_SUB_TYPE
|
|
|
|
|
|
def test_binding_accepts_custom_mapping_go_path_but_not_its_a_suffix():
|
|
binding = task_deeplink_binding(
|
|
{
|
|
"taskId": "task-1",
|
|
"taskRecordId": "task-1",
|
|
"taskGoPath": "区域研判",
|
|
}
|
|
)
|
|
|
|
assert binding is not None
|
|
assert binding["go_path"] == "区域研判"
|
|
assert task_deeplink_binding(
|
|
{
|
|
"taskId": "task-1",
|
|
"taskRecordId": "task-1",
|
|
"taskGoPath": "区域研判-A",
|
|
}
|
|
) is None
|