"""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/knowledge/list", ] 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", "/api/v1/knowledge/upload", ] PROTECTED_DELETE = [ "/api/v1/knowledge/1", ] 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", PROTECTED_DELETE) async def test_protected_delete_without_token_is_unauthorized(path: str) -> None: """DELETE 没有请求体可校,所以未带令牌必须是明确的 401(知识库删除端点的闸门)。""" response = await send("DELETE", path) assert response.status_code == 401, f"DELETE {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 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