318 lines
12 KiB
Python
318 lines
12 KiB
Python
"""b5-1 task-detail persistence and API contract tests."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
from fastapi import FastAPI, HTTPException
|
||
from fastapi.testclient import TestClient
|
||
from pydantic import ValidationError
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
from starlette.requests import Request
|
||
|
||
from app.gateway import auth_middleware, csrf_middleware
|
||
from app.gateway.routers import task_deeplink_details as router
|
||
from deerflow.persistence.task_deeplink_details import (
|
||
MemoryTaskDeeplinkDetailStore,
|
||
TaskDeeplinkDetailRepository,
|
||
)
|
||
from deerflow.persistence.task_deeplink_details.model import TaskDeeplinkDetailRow
|
||
|
||
|
||
def _request(store, *, role: str = "admin", external: bool = False, anonymous: bool = False):
|
||
state = SimpleNamespace(
|
||
user=None
|
||
if anonymous
|
||
else SimpleNamespace(id="admin-1", system_role=role)
|
||
)
|
||
if external:
|
||
state.task_deeplink_external_writer = True
|
||
return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(task_deeplink_detail_store=store)), state=state)
|
||
|
||
|
||
def _http_request(method: str, path: str, headers: dict[str, str] | None = None) -> Request:
|
||
return Request(
|
||
{
|
||
"type": "http",
|
||
"http_version": "1.1",
|
||
"method": method,
|
||
"scheme": "http",
|
||
"path": path,
|
||
"query_string": b"",
|
||
"headers": [
|
||
(key.lower().encode("latin-1"), value.encode("latin-1"))
|
||
for key, value in (headers or {}).items()
|
||
],
|
||
"client": ("test", 1234),
|
||
"server": ("test", 80),
|
||
}
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_memory_store_skips_identical_snapshot_writes_and_versions_changes() -> None:
|
||
store = MemoryTaskDeeplinkDetailStore()
|
||
first, created, changed = await store.upsert_detail(
|
||
"b5-1",
|
||
"task-1",
|
||
task_name="任务甲",
|
||
task_content="第一版详情",
|
||
overview="概述",
|
||
extra={"source": "external"},
|
||
updated_by="tester",
|
||
)
|
||
assert created is True
|
||
assert changed is True
|
||
assert first["version"] == 1
|
||
|
||
second, created, changed = await store.upsert_detail(
|
||
"b5-1",
|
||
"task-1",
|
||
task_name="任务甲",
|
||
task_content="第二版详情",
|
||
overview="",
|
||
extra={"priority": 3},
|
||
updated_by="tester-2",
|
||
)
|
||
assert created is False
|
||
assert changed is True
|
||
assert second["version"] == 2
|
||
assert second["extra"] == {"priority": 3}
|
||
|
||
same, created, changed = await store.upsert_detail(
|
||
"b5-1",
|
||
"task-1",
|
||
task_name="任务甲",
|
||
task_content="第二版详情",
|
||
overview="",
|
||
extra={"priority": 3},
|
||
updated_by="retry-writer",
|
||
)
|
||
assert created is False
|
||
assert changed is False
|
||
assert same["version"] == 2
|
||
assert same["updated_at"] == second["updated_at"]
|
||
assert await store.get_detail("b5-1", "task-2") is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_memory_store_partitions_same_id_by_sub_type() -> None:
|
||
store = MemoryTaskDeeplinkDetailStore()
|
||
_, created_a, _ = await store.upsert_detail(
|
||
"b5-1",
|
||
"task-9",
|
||
sub_type="yuqing",
|
||
task_name="舆情任务",
|
||
task_content="舆情详情",
|
||
overview="",
|
||
extra={},
|
||
updated_by="writer",
|
||
)
|
||
_, created_b, _ = await store.upsert_detail(
|
||
"b5-1",
|
||
"task-9",
|
||
sub_type="other",
|
||
task_name="其它任务",
|
||
task_content="其它详情",
|
||
overview="",
|
||
extra={},
|
||
updated_by="writer",
|
||
)
|
||
assert created_a is True and created_b is True
|
||
|
||
yuqing = await store.get_detail("b5-1", "task-9", sub_type="yuqing")
|
||
other = await store.get_detail("b5-1", "task-9", sub_type="other")
|
||
untyped = await store.get_detail("b5-1", "task-9")
|
||
assert yuqing is not None and yuqing["task_name"] == "舆情任务" and yuqing["sub_type"] == "yuqing"
|
||
assert other is not None and other["task_name"] == "其它任务"
|
||
assert untyped is None
|
||
|
||
# 同类型重试幂等；跨类型互不覆盖。
|
||
_, created, changed = await store.upsert_detail(
|
||
"b5-1",
|
||
"task-9",
|
||
sub_type="yuqing",
|
||
task_name="舆情任务",
|
||
task_content="舆情详情",
|
||
overview="",
|
||
extra={},
|
||
updated_by="retry",
|
||
)
|
||
assert (created, changed) == (False, False)
|
||
assert (await store.get_detail("b5-1", "task-9", sub_type="other"))["version"] == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_sql_store_round_trips_detail_snapshot() -> None:
|
||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
try:
|
||
async with engine.begin() as connection:
|
||
await connection.run_sync(TaskDeeplinkDetailRow.__table__.create)
|
||
store = TaskDeeplinkDetailRepository(async_sessionmaker(engine, expire_on_commit=False))
|
||
saved, created, changed = await store.upsert_detail(
|
||
"b5-1",
|
||
"task-42",
|
||
task_name="SQL 任务",
|
||
task_content="详情",
|
||
overview="概述",
|
||
extra={"tags": ["a", "b"]},
|
||
updated_by="admin",
|
||
)
|
||
assert created is True
|
||
assert changed is True
|
||
assert saved["detail_id"] == "task-42"
|
||
assert (await store.get_detail("b5-1", "task-42"))["extra"] == {"tags": ["a", "b"]}
|
||
|
||
same, created, changed = await store.upsert_detail(
|
||
"b5-1",
|
||
"task-42",
|
||
task_name="SQL 任务",
|
||
task_content="详情",
|
||
overview="概述",
|
||
extra={"tags": ["a", "b"]},
|
||
updated_by="retry-writer",
|
||
)
|
||
assert created is False
|
||
assert changed is False
|
||
assert same["version"] == 1
|
||
finally:
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_router_returns_camel_case_contract_and_allows_open_writes() -> None:
|
||
store = MemoryTaskDeeplinkDetailStore()
|
||
request = _request(store)
|
||
body = router.TaskDeeplinkDetailUpsertRequest.model_validate(
|
||
{
|
||
"taskName": "任务分析名称",
|
||
"taskContent": "这是任务详情",
|
||
"overview": "任务概述",
|
||
"extra": {"origin": "b5", "severity": 2},
|
||
}
|
||
)
|
||
saved = await router.upsert_task_deeplink_detail(request, body, "b5-1", "id-7")
|
||
assert saved.created is True
|
||
assert saved.changed is True
|
||
assert saved.data.task_name == "任务分析名称"
|
||
assert saved.data.extra == {"origin": "b5", "severity": 2}
|
||
|
||
fetched = await router.get_task_deeplink_detail(request, "b5-1", "id-7", type="")
|
||
assert fetched.data.detail_id == "id-7"
|
||
assert fetched.data.model_dump(by_alias=True)["taskName"] == "任务分析名称"
|
||
assert fetched.data.model_dump(by_alias=True)["type"] == "b5-1"
|
||
assert fetched.data.model_dump(by_alias=True)["taskType"] == ""
|
||
|
||
retry = await router.upsert_task_deeplink_detail(request, body, "b5-1", "id-7")
|
||
assert retry.created is False
|
||
assert retry.changed is False
|
||
assert retry.data.version == 1
|
||
|
||
# 写入端点对所有人开放：普通用户与未登录调用方都可直接写。
|
||
by_user = await router.upsert_task_deeplink_detail(
|
||
_request(store, role="user"), body, "b5-1", "id-8"
|
||
)
|
||
assert by_user.created is True
|
||
|
||
by_anonymous = await router.upsert_task_deeplink_detail(
|
||
_request(store, anonymous=True), body, "b5-1", "id-anonymous"
|
||
)
|
||
assert by_anonymous.created is True
|
||
stored = await store.get_detail("b5-1", "id-anonymous")
|
||
assert stored is not None and stored["updated_by"] == "external:task-deeplink"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_router_partitions_same_id_by_request_type() -> None:
|
||
store = MemoryTaskDeeplinkDetailStore()
|
||
request = _request(store)
|
||
|
||
typed_body = router.TaskDeeplinkDetailUpsertRequest.model_validate(
|
||
{"taskName": "舆情任务", "taskContent": "舆情详情", "type": "yuqing"}
|
||
)
|
||
saved = await router.upsert_task_deeplink_detail(request, typed_body, "b5-1", "task-x")
|
||
assert saved.created is True
|
||
assert saved.data.detail_sub_type == "yuqing"
|
||
assert saved.data.model_dump(by_alias=True)["taskType"] == "yuqing"
|
||
|
||
other_body = router.TaskDeeplinkDetailUpsertRequest.model_validate(
|
||
{"taskName": "行动任务", "taskContent": "行动详情", "type": "action"}
|
||
)
|
||
await router.upsert_task_deeplink_detail(request, other_body, "b5-1", "task-x")
|
||
|
||
fetched = await router.get_task_deeplink_detail(request, "b5-1", "task-x", type="yuqing")
|
||
assert fetched.data.task_name == "舆情任务"
|
||
fetched_other = await router.get_task_deeplink_detail(request, "b5-1", "task-x", type="action")
|
||
assert fetched_other.data.task_name == "行动任务"
|
||
|
||
# 该 id 未写无类型分区：不带 ?type= 读不到（不会回落到任意类型）。
|
||
with pytest.raises(HTTPException) as error:
|
||
await router.get_task_deeplink_detail(request, "b5-1", "task-x", type="")
|
||
assert error.value.status_code == 404
|
||
|
||
|
||
def test_router_rejects_blank_required_fields_and_non_object_extra() -> None:
|
||
with pytest.raises(ValidationError):
|
||
router.TaskDeeplinkDetailUpsertRequest.model_validate(
|
||
{"taskName": " ", "taskContent": "详情"}
|
||
)
|
||
with pytest.raises(ValidationError):
|
||
router.TaskDeeplinkDetailUpsertRequest.model_validate(
|
||
{"taskName": "名称", "taskContent": "详情", "extra": ["not", "an", "object"]}
|
||
)
|
||
|
||
|
||
def test_b5_put_is_credential_free_and_bypasses_csrf() -> None:
|
||
# 无 key、无登录态：PUT 直接放行并打上外部写入标记（供审计）。
|
||
bare = _http_request("PUT", "/api/task-deeplink-details/b5-1/id-9")
|
||
assert auth_middleware._is_task_deeplink_external_write(bare, bare.url.path)
|
||
assert auth_middleware._authorize_task_deeplink_external_write(bare) is None
|
||
assert bare.state.task_deeplink_external_writer is True
|
||
assert csrf_middleware.should_check_csrf(bare) is False
|
||
|
||
# 旧集成方带的 X-Integration-Key 不再校验（只记指纹），错误 key 也放行。
|
||
stale_key = _http_request(
|
||
"PUT",
|
||
"/api/task-deeplink-details/b5-1/id-9",
|
||
{"X-Integration-Key": "anything"},
|
||
)
|
||
assert auth_middleware._authorize_task_deeplink_external_write(stale_key) is None
|
||
assert stale_key.state.task_deeplink_external_key_fingerprint
|
||
|
||
# 通道只覆盖 PUT；其它方法仍走正常登录校验。
|
||
post_request = _http_request("POST", "/api/task-deeplink-details/b5-1/id-9")
|
||
assert not auth_middleware._is_task_deeplink_external_write(post_request, post_request.url.path)
|
||
assert csrf_middleware.should_check_csrf(post_request) is True
|
||
|
||
|
||
def test_unauthenticated_put_succeeds_but_read_still_requires_login() -> None:
|
||
app = FastAPI()
|
||
app.state.task_deeplink_detail_store = MemoryTaskDeeplinkDetailStore()
|
||
app.include_router(router.router)
|
||
app.add_middleware(auth_middleware.AuthMiddleware)
|
||
|
||
client = TestClient(app)
|
||
written = client.put(
|
||
"/api/task-deeplink-details/b5-1/task-open",
|
||
json={"taskName": "开放写入任务", "taskContent": "这是任务详情"},
|
||
)
|
||
assert written.status_code == 200
|
||
assert written.json()["created"] is True
|
||
|
||
# 同一 id 再写一个带类型的快照，互不覆盖（端到端）。
|
||
typed = client.put(
|
||
"/api/task-deeplink-details/b5-1/task-open",
|
||
json={"taskName": "同id不同类型", "taskContent": "类型详情", "type": "yuqing"},
|
||
)
|
||
assert typed.status_code == 200
|
||
assert typed.json()["data"]["taskType"] == "yuqing"
|
||
assert typed.json()["created"] is True
|
||
|
||
# 读取仍走正常登录校验：未登录 GET（无论带不带 type）→ 401，
|
||
# 不因写入开放而连带放开。
|
||
denied = client.get("/api/task-deeplink-details/b5-1/task-open")
|
||
assert denied.status_code == 401
|
||
denied_typed = client.get("/api/task-deeplink-details/b5-1/task-open?type=yuqing")
|
||
assert denied_typed.status_code == 401
|