e082bd2d98
- 新增`api_response.py`统一响应封装工具类，提供标准成功/错误响应构造方法 - 重构WonderQ-Admin全局异常处理器，将所有异常转换为标准响应格式 - 修改所有公共和管理端接口的返回逻辑，统一使用`code`（与HTTP状态码一致）、`msg`和`data`的三层结构 - 新增`api-response-contract.md`文档，定义完整的三端统一JSON响应规范 - 更新所有领域API文档，明确业务数据需位于`data`字段内，补充响应格式说明 - 为WonderQ-MiniAPP和WonderQ-Admin-UI新增响应解析逻辑和类型定义，自动完成协议校验和错误处理 - 更新所有测试用例，适配新的响应结构确保接口符合契约要求 - 新增`module-config-api.md`模块配置API文档，补充站点模块配置的接口约定 - 更新项目README文档，调整文档分类顺序将响应契约置于首位
331 lines
10 KiB
Python
331 lines
10 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import ValidationError
|
|
|
|
from app.database import get_db
|
|
from app.main import create_app
|
|
from app.models import AdminUser, WanfaCategory, WanfaRoute
|
|
from app.auth import require_admin
|
|
from app.schemas import WanfaRouteCreate
|
|
|
|
|
|
class FakeScalarResult:
|
|
def __init__(self, items):
|
|
self.items = items
|
|
|
|
def unique(self):
|
|
return self
|
|
|
|
def all(self):
|
|
return self.items
|
|
|
|
|
|
class FakeDb:
|
|
def __init__(self, *, scalar_results=None, scalar_values=None, get_result=None, get_results=None):
|
|
self.scalar_results = list(scalar_results or [])
|
|
self.scalar_values = list(scalar_values or [])
|
|
self.get_result = get_result
|
|
self.get_results = get_results or {}
|
|
self.added = []
|
|
self.deleted = []
|
|
self.committed = False
|
|
|
|
def scalars(self, _statement):
|
|
return FakeScalarResult(self.scalar_results.pop(0) if self.scalar_results else [])
|
|
|
|
def scalar(self, _statement):
|
|
return self.scalar_values.pop(0) if self.scalar_values else None
|
|
|
|
def get(self, model, item_id):
|
|
return self.get_results.get((model, item_id), self.get_result)
|
|
|
|
def add(self, item):
|
|
self.added.append(item)
|
|
|
|
def delete(self, item):
|
|
self.deleted.append(item)
|
|
|
|
def flush(self):
|
|
for item in self.added:
|
|
if not getattr(item, "id", None):
|
|
item.id = f"{item.__class__.__name__.lower()}-test-id"
|
|
|
|
def commit(self):
|
|
self.committed = True
|
|
|
|
|
|
def make_admin_user():
|
|
return AdminUser(
|
|
id="admin-test",
|
|
email="admin@example.test",
|
|
name="Admin",
|
|
role="admin",
|
|
passwordHash="not-used",
|
|
isActive=True,
|
|
)
|
|
|
|
|
|
def authenticated_app(fake_db):
|
|
app = create_app()
|
|
app.dependency_overrides[get_db] = lambda: fake_db
|
|
app.dependency_overrides[require_admin] = make_admin_user
|
|
return app
|
|
|
|
|
|
def make_route(**overrides):
|
|
return WanfaRoute(
|
|
id=overrides.get("id", "family-water"),
|
|
categoryId=overrides.get("categoryId", "family-route"),
|
|
title=overrides.get("title", "亲子玩水"),
|
|
subtitle=overrides.get("subtitle", "贵州·轻松节奏与自然课堂"),
|
|
image=overrides.get("image", "https://example.test/family-water.jpg"),
|
|
routeCount=overrides.get("routeCount", 4),
|
|
demandKeyword=overrides.get("demandKeyword", "亲子玩水"),
|
|
sortOrder=overrides.get("sortOrder", 0),
|
|
createdAt=datetime(2026, 1, 1),
|
|
updatedAt=datetime(2026, 1, 2),
|
|
)
|
|
|
|
|
|
def make_category(*, routes=None, **overrides):
|
|
return WanfaCategory(
|
|
id=overrides.get("id", "family-route"),
|
|
label=overrides.get("label", "亲子路线"),
|
|
sortOrder=overrides.get("sortOrder", 0),
|
|
createdAt=datetime(2026, 1, 1),
|
|
updatedAt=datetime(2026, 1, 2),
|
|
routes=list(routes or []),
|
|
)
|
|
|
|
|
|
def test_wanfa_route_schema_trims_required_strings():
|
|
route = WanfaRouteCreate(
|
|
title=" 亲子玩水 ",
|
|
subtitle=" 贵州·轻松节奏与自然课堂 ",
|
|
image="https://example.test/family-water.jpg",
|
|
routeCount=4,
|
|
demandKeyword=" 亲子玩水 ",
|
|
)
|
|
|
|
assert route.title == "亲子玩水"
|
|
assert route.subtitle == "贵州·轻松节奏与自然课堂"
|
|
assert route.demandKeyword == "亲子玩水"
|
|
|
|
|
|
def test_wanfa_route_schema_rejects_empty_required_strings():
|
|
with pytest.raises(ValidationError):
|
|
WanfaRouteCreate(
|
|
title="",
|
|
subtitle="路线说明",
|
|
image="https://example.test/family-water.jpg",
|
|
routeCount=0,
|
|
demandKeyword="亲子玩水",
|
|
)
|
|
|
|
|
|
def test_admin_wanfa_list_returns_nested_categories_and_routes():
|
|
route = make_route()
|
|
category = make_category(routes=[route])
|
|
fake_db = FakeDb(scalar_results=[[category]])
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).get("/api/admin/wanfa/categories")
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"code": 200,
|
|
"msg": "success",
|
|
"data": {
|
|
"categories": [
|
|
{
|
|
"id": "family-route",
|
|
"label": "亲子路线",
|
|
"routes": [
|
|
{
|
|
"id": "family-water",
|
|
"title": "亲子玩水",
|
|
"subtitle": "贵州·轻松节奏与自然课堂",
|
|
"image": "https://example.test/family-water.jpg",
|
|
"routeCount": 4,
|
|
"demandKeyword": "亲子玩水",
|
|
}
|
|
],
|
|
}
|
|
]
|
|
},
|
|
}
|
|
|
|
|
|
def test_admin_wanfa_create_category_appends_to_the_end():
|
|
fake_db = FakeDb(scalar_values=[2])
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).post("/api/admin/wanfa/categories", json={"label": " 新玩法 "})
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["data"]["label"] == "新玩法"
|
|
assert fake_db.added[0].label == "新玩法"
|
|
assert fake_db.added[0].sortOrder == 3
|
|
assert fake_db.committed
|
|
|
|
|
|
def test_admin_wanfa_create_route_uses_category_sort_order():
|
|
category = make_category(routes=[])
|
|
fake_db = FakeDb(get_result=category, scalar_values=[1])
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).post(
|
|
"/api/admin/wanfa/categories/family-route/routes",
|
|
json={
|
|
"title": "亲子溯溪",
|
|
"subtitle": "贵州·溪谷轻徒步",
|
|
"image": "https://example.test/family-stream.jpg",
|
|
"routeCount": 2,
|
|
"demandKeyword": "亲子溯溪",
|
|
},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["data"]["id"] == "wanfaroute-test-id"
|
|
assert fake_db.added[0].categoryId == "family-route"
|
|
assert fake_db.added[0].sortOrder == 2
|
|
|
|
|
|
def test_admin_wanfa_delete_rejects_non_empty_category():
|
|
category = make_category(routes=[make_route()])
|
|
fake_db = FakeDb(get_result=category)
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).delete("/api/admin/wanfa/categories/family-route")
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 409
|
|
assert response.json()["errorCode"] == "WANFA_CATEGORY_NOT_EMPTY"
|
|
assert not fake_db.deleted
|
|
assert not fake_db.committed
|
|
|
|
|
|
def test_admin_wanfa_delete_rejects_category_linked_to_home_recommendation():
|
|
category = make_category(routes=[])
|
|
fake_db = FakeDb(get_result=category, scalar_values=[1])
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).delete("/api/admin/wanfa/categories/family-route")
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 409
|
|
assert response.json()["errorCode"] == "WANFA_CATEGORY_RECOMMENDED"
|
|
assert not fake_db.deleted
|
|
assert not fake_db.committed
|
|
|
|
|
|
def test_admin_wanfa_route_update_returns_updated_route():
|
|
route = make_route()
|
|
category = make_category()
|
|
fake_db = FakeDb(
|
|
get_results={
|
|
(WanfaCategory, "family-route"): category,
|
|
(WanfaRoute, "family-water"): route,
|
|
}
|
|
)
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).patch(
|
|
"/api/admin/wanfa/categories/family-route/routes/family-water",
|
|
json={"title": "亲子溯溪", "routeCount": 5},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["data"]["title"] == "亲子溯溪"
|
|
assert response.json()["data"]["routeCount"] == 5
|
|
assert route.title == "亲子溯溪"
|
|
assert route.routeCount == 5
|
|
assert fake_db.committed
|
|
|
|
|
|
def test_admin_wanfa_route_delete_returns_route_id():
|
|
route = make_route()
|
|
category = make_category(routes=[route])
|
|
fake_db = FakeDb(
|
|
get_results={
|
|
(WanfaCategory, "family-route"): category,
|
|
(WanfaRoute, "family-water"): route,
|
|
}
|
|
)
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).delete(
|
|
"/api/admin/wanfa/categories/family-route/routes/family-water"
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"code": 200,
|
|
"msg": "success",
|
|
"data": {"id": "family-water"},
|
|
}
|
|
assert fake_db.deleted == [route]
|
|
assert fake_db.committed
|
|
|
|
|
|
def test_admin_wanfa_category_reorder_rejects_incomplete_ids():
|
|
categories = [make_category(id="family-route"), make_category(id="photo-route", label="旅拍路线")]
|
|
fake_db = FakeDb(scalar_results=[categories])
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).patch(
|
|
"/api/admin/wanfa/categories/reorder",
|
|
json={"itemIds": ["family-route"]},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["errorCode"] == "WANFA_CATEGORY_REORDER_INVALID"
|
|
assert not fake_db.committed
|
|
|
|
|
|
def test_admin_wanfa_route_reorder_reassigns_sort_order():
|
|
routes = [
|
|
make_route(id="route-1", sortOrder=0),
|
|
make_route(id="route-2", sortOrder=1),
|
|
]
|
|
category = make_category(routes=routes)
|
|
fake_db = FakeDb(get_result=category)
|
|
app = authenticated_app(fake_db)
|
|
|
|
try:
|
|
response = TestClient(app).patch(
|
|
"/api/admin/wanfa/categories/family-route/routes/reorder",
|
|
json={"itemIds": ["route-2", "route-1"]},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert response.status_code == 200
|
|
assert [item["id"] for item in response.json()["data"]["items"]] == ["route-2", "route-1"]
|
|
assert {route.id: route.sortOrder for route in routes} == {"route-2": 0, "route-1": 1}
|
|
assert fake_db.committed
|