Files
WonderQ-Project/WonderQ-Admin/tests/test_details.py
duanshuwen e082bd2d98 feat(api): 实现三端统一的JSON API响应契约
- 新增`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文档,调整文档分类顺序将响应契约置于首位
2026-08-19 22:02:20 +08:00

230 lines
6.9 KiB
Python

from datetime import datetime
from fastapi.testclient import TestClient
from pydantic import ValidationError
import pytest
from app.auth import require_admin
from app.database import get_db
from app.main import create_app
from app.models import AdminUser, DetailRecord
from app.schemas import DetailCreate
class FakeScalarResult:
def __init__(self, items):
self.items = items
def all(self):
return self.items
class FakeDb:
def __init__(self, *, scalar_results=None, scalar_values=None, get_results=None):
self.scalar_results = list(scalar_results or [])
self.scalar_values = list(scalar_values or [])
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))
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 = "detail-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 make_detail(**overrides):
values = {
"id": "detail-family-water",
"key": "family-water",
"eyebrow": "玩法推荐",
"duration": "5天4晚",
"title": "亲子玩水",
"subtitle": "贵州·轻松节奏与自然课堂",
"intro": "沿着贵州山地的自然纹理深入探索。",
"highlights": ["核心景观串联"],
"included": ["行程内用车与接送服务"],
"excluded": ["往返大交通"],
"notes": ["请准备轻便雨具。"],
"gallery": ["https://example.test/family-water.jpg"],
"isActive": True,
"sortOrder": 0,
"createdAt": datetime(2026, 1, 1),
"updatedAt": datetime(2026, 1, 2),
}
values.update(overrides)
return DetailRecord(**values)
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 test_detail_schema_trims_text_and_array_items():
detail = DetailCreate(
key=" family-water ",
eyebrow=" 玩法推荐 ",
duration=" 5天4晚 ",
title=" 亲子玩水 ",
subtitle=" 贵州·轻松节奏与自然课堂 ",
intro=" 沿着山地深入探索。 ",
highlights=[" 核心景观串联 ", "", " 小团出行 "],
included=[" 行程内用车 "],
excluded=[" 往返大交通 "],
notes=[" 请准备雨具。 "],
gallery=[" https://example.test/family-water.jpg "],
)
assert detail.key == "family-water"
assert detail.highlights == ["核心景观串联", "小团出行"]
assert detail.gallery == ["https://example.test/family-water.jpg"]
def test_detail_schema_rejects_invalid_gallery_url():
with pytest.raises(ValidationError):
DetailCreate(
key="family-water",
eyebrow="玩法推荐",
duration="5天4晚",
title="亲子玩水",
subtitle="贵州路线",
intro="介绍",
highlights=["亮点"],
included=["包含"],
excluded=["不含"],
notes=["注意"],
gallery=["not-a-url"],
)
def test_public_detail_returns_frontend_contract_without_auth():
fake_db = FakeDb(scalar_values=[make_detail()])
app = create_app()
app.dependency_overrides[get_db] = lambda: fake_db
try:
response = TestClient(app).get("/api/public/details/family-water")
finally:
app.dependency_overrides.clear()
assert response.status_code == 200
assert response.json() == {
"code": 200,
"msg": "success",
"data": {
"key": "family-water",
"eyebrow": "玩法推荐",
"duration": "5天4晚",
"title": "亲子玩水",
"subtitle": "贵州·轻松节奏与自然课堂",
"intro": "沿着贵州山地的自然纹理深入探索。",
"highlights": ["核心景观串联"],
"included": ["行程内用车与接送服务"],
"excluded": ["往返大交通"],
"notes": ["请准备轻便雨具。"],
"gallery": ["https://example.test/family-water.jpg"],
},
}
def test_public_detail_returns_not_found_for_missing_key():
fake_db = FakeDb(scalar_values=[None])
app = create_app()
app.dependency_overrides[get_db] = lambda: fake_db
try:
response = TestClient(app).get("/api/public/details/missing")
finally:
app.dependency_overrides.clear()
assert response.status_code == 404
def test_public_detail_returns_not_found_for_inactive_record():
# The query filters inactive rows, so an inactive match is represented as no scalar result.
fake_db = FakeDb(scalar_values=[None])
app = create_app()
app.dependency_overrides[get_db] = lambda: fake_db
try:
response = TestClient(app).get("/api/public/details/family-water")
finally:
app.dependency_overrides.clear()
assert response.status_code == 404
def test_admin_detail_list_includes_inactive_records():
fake_db = FakeDb(scalar_results=[[make_detail(isActive=False)]])
app = authenticated_app(fake_db)
try:
response = TestClient(app).get("/api/admin/details")
finally:
app.dependency_overrides.clear()
assert response.status_code == 200
assert response.json()["data"]["details"][0]["isActive"] is False
def test_admin_detail_create_normalizes_key_and_appends():
fake_db = FakeDb(scalar_values=[None, 2])
app = authenticated_app(fake_db)
try:
response = TestClient(app).post(
"/api/admin/details",
json={
"key": " family-water ",
"eyebrow": "玩法推荐",
"duration": "5天4晚",
"title": "亲子玩水",
"subtitle": "贵州路线",
"intro": "沿着山地深入探索。",
"highlights": ["核心景观串联"],
"included": [],
"excluded": [],
"notes": [],
"gallery": ["https://example.test/family-water.jpg"],
},
)
finally:
app.dependency_overrides.clear()
assert response.status_code == 201
assert response.json()["data"]["key"] == "family-water"
assert fake_db.added[0].sortOrder == 3