94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
|
"""T 段（账户看板 + 场内模拟交易）Controller 路由契约测试。
|
||
|
|
|
||
|
|
不连数据库。覆盖：
|
||
|
|
|
||
|
|
- 9 个端点路由**已注册**且未授权时不能到达业务逻辑（401 / 422）；
|
||
|
|
- 端点路径和 HTTP 方法匹配 §19 文档（避免注册漂移）。
|
||
|
|
|
||
|
|
受保护路由加入 `tests/unit/api/test_controller_routing_contract.py` 同款断言。
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.main import create_app
|
||
|
|
|
||
|
|
# ---- 路由清单（与 docs/05 §19 T 段一一对应） ----
|
||
|
|
|
||
|
|
T_GET_LIST = [
|
||
|
|
"/api/v1/users/me/account/dashboard", # T001
|
||
|
|
"/api/v1/users/me/orders", # T003
|
||
|
|
"/api/v1/users/me/holdings", # T006
|
||
|
|
"/api/v1/users/me/transactions", # T007
|
||
|
|
"/api/v1/users/me/cash-ledger", # T009
|
||
|
|
]
|
||
|
|
T_GET_DETAIL = [
|
||
|
|
"/api/v1/users/me/orders/SO2026010100000000000", # T004
|
||
|
|
"/api/v1/users/me/transactions/TX2026010100000000000", # T008
|
||
|
|
]
|
||
|
|
T_POST = [
|
||
|
|
"/api/v1/users/me/orders", # T002
|
||
|
|
"/api/v1/users/me/orders/SO2026010100000000000/cancellations", # T005
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
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", T_GET_LIST)
|
||
|
|
async def test_t_list_endpoint_is_registered_and_protected(path: str) -> None:
|
||
|
|
"""T 段列表端点路由必须存在；缺 token 时 401 而不是 404。"""
|
||
|
|
response = await send("GET", path)
|
||
|
|
assert response.status_code == 401, (
|
||
|
|
f"GET {path} 未授权应 401，实际 {response.status_code}（路由可能漏注册）"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("path", T_GET_DETAIL)
|
||
|
|
async def test_t_detail_endpoint_is_registered_and_protected(path: str) -> None:
|
||
|
|
response = await send("GET", path)
|
||
|
|
assert response.status_code == 401, f"GET {path} -> {response.status_code}"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("path", T_POST)
|
||
|
|
async def test_t_post_endpoint_is_registered_and_protected(path: str) -> None:
|
||
|
|
"""T 段 POST / POST * 端点：未带令牌永远不能成功（401/422 之一）。"""
|
||
|
|
response = await send("POST", path)
|
||
|
|
assert response.status_code in {401, 422}, f"POST {path} -> {response.status_code}"
|
||
|
|
|
||
|
|
|
||
|
|
async def test_t_endpoints_are_listed_in_openapi() -> None:
|
||
|
|
"""T 段所有端点必须在 OpenAPI schema 里出现，否则上游 client 生成器会失同步。"""
|
||
|
|
app = create_app()
|
||
|
|
paths = app.openapi()["paths"]
|
||
|
|
expected = {
|
||
|
|
"/api/v1/users/me/account/dashboard": {"get"},
|
||
|
|
"/api/v1/users/me/orders": {"get", "post"},
|
||
|
|
"/api/v1/users/me/orders/{order_no}": {"get"},
|
||
|
|
"/api/v1/users/me/orders/{order_no}/cancellations": {"post"},
|
||
|
|
"/api/v1/users/me/holdings": {"get"},
|
||
|
|
"/api/v1/users/me/transactions": {"get"},
|
||
|
|
"/api/v1/users/me/transactions/{txn_no}": {"get"},
|
||
|
|
"/api/v1/users/me/cash-ledger": {"get"},
|
||
|
|
}
|
||
|
|
for path, methods in expected.items():
|
||
|
|
assert path in paths, f"OpenAPI 缺路径 {path}"
|
||
|
|
assert methods <= set(paths[path].keys()), (
|
||
|
|
f"OpenAPI {path} 方法集合应为 {methods}，实际 {set(paths[path].keys())}"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
async def test_t_envelope_shape_is_preserved() -> None:
|
||
|
|
"""T 段响应必须使用 §3.3 信封 ``{data, meta}``。
|
||
|
|
|
||
|
|
即便未授权也保持信封一致性以便客户端统一处理。
|
||
|
|
"""
|
||
|
|
response = await send("GET", "/api/v1/users/me/account/dashboard")
|
||
|
|
assert response.status_code == 401
|