98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
"""Controller 路由契约测试：鉴权闸门与路由注册。
|
||
|
||
两类断言：
|
||
1. **受保护路由在缺少 `Authorization` 时不得成功**——GET 必须明确 401；POST 因为 FastAPI
|
||
先校验请求体，无 body 时会得到 422，因此断言"不是 2xx 且是 401/422 之一"，重点是
|
||
**未授权不能拿到成功响应**，而不是具体哪一个码。
|
||
2. 公开运维路由可达、未注册路径返回 404——防止路由注册被改错却无人发现。
|
||
|
||
全部用 `httpx.ASGITransport` 进程内调用，不连数据库（鉴权在依赖层就返回）。
|
||
"""
|
||
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from app.main import create_app
|
||
|
||
PROTECTED_GET = [
|
||
"/api/v1/agent-runs/run-x",
|
||
"/api/v1/agent-runs/run-x/events",
|
||
"/api/v1/conversations/session-1",
|
||
"/api/v1/conversations/session-1/messages",
|
||
"/api/v1/handover-requests/1",
|
||
"/api/v1/knowledge-references/token-abcdefghijklmnopqrst",
|
||
"/api/v1/users/me/memory-profile",
|
||
"/api/v1/admin/config-releases",
|
||
"/api/v1/admin/customer-service/handover-tickets",
|
||
"/api/v1/admin/customer-service/handover-tickets/ticket-x",
|
||
]
|
||
|
||
PROTECTED_POST = [
|
||
"/api/v1/agent-runs",
|
||
"/api/v1/conversations",
|
||
"/api/v1/conversations/session-1/closures",
|
||
"/api/v1/conversations/session-1/handover-requests",
|
||
"/api/v1/agent-runs/run-x/cancellations",
|
||
"/api/v1/conversation-messages/1/feedback",
|
||
]
|
||
|
||
PUBLIC_GET = ["/internal/health/live"]
|
||
|
||
|
||
async def send(method: str, path: str) -> httpx.Response:
|
||
app = create_app()
|
||
transport = httpx.ASGITransport(app=app)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
return await client.request(method, path)
|
||
|
||
|
||
@pytest.mark.parametrize("path", PROTECTED_GET)
|
||
async def test_protected_get_without_token_is_unauthorized(path: str) -> None:
|
||
response = await send("GET", path)
|
||
|
||
assert response.status_code == 401, f"GET {path} -> {response.status_code}"
|
||
|
||
|
||
@pytest.mark.parametrize("path", PROTECTED_POST)
|
||
async def test_protected_post_without_token_never_succeeds(path: str) -> None:
|
||
response = await send("POST", path)
|
||
|
||
assert response.status_code in {401, 422}, f"POST {path} -> {response.status_code}"
|
||
|
||
|
||
@pytest.mark.parametrize("path", PUBLIC_GET)
|
||
async def test_public_operational_route_is_reachable(path: str) -> None:
|
||
response = await send("GET", path)
|
||
|
||
assert response.status_code == 200, f"GET {path} -> {response.status_code}"
|
||
|
||
|
||
async def test_unknown_path_is_not_found() -> None:
|
||
response = await send("GET", "/api/v1/definitely-not-a-route")
|
||
|
||
assert response.status_code == 404
|
||
|
||
|
||
def test_app_registers_platform_and_offsite_routes() -> None:
|
||
"""整合后的应用必须同时保留新底座平台路由和场外基金业务路由。"""
|
||
paths = {
|
||
child.path
|
||
for route in create_app().routes
|
||
if hasattr(route, "original_router")
|
||
for child in route.original_router.routes
|
||
if hasattr(child, "path")
|
||
}
|
||
|
||
assert "/api/v1/agent-runs" in paths
|
||
assert any(path.startswith("/api/v1/offsite-fund") for path in paths)
|
||
|
||
|
||
async def test_unauthorized_envelope_shape() -> None:
|
||
"""401 的错误信封必须与文档一致，否则客户端无法统一处理。"""
|
||
response = await send("GET", "/api/v1/agent-runs/run-x")
|
||
|
||
body: Any = response.json()
|
||
assert "detail" in body or "error" in body
|