feat: Add WonderQ-MiniAPP Public API documentation
- Introduced a comprehensive API contract for the WonderQ-MiniAPP, detailing endpoints for site configuration, product listings, and lead submissions. - Defined data types for various entities including HeroSlide, Destination, Theme, CtaBanner, PublicProduct, and more. - Specified request and response formats, including error handling guidelines. chore: Update requirements to include python-multipart - Added python-multipart dependency to requirements.txt for handling file uploads. test: Implement API contract tests - Created test suite for API contracts, validating serializers and endpoints for public products and leads. - Included tests for destination and product serializers, ensuring correct data handling and validation. test: Add configuration tests for OSS settings - Implemented tests to verify that OSS settings are correctly loaded from environment variables.
This commit is contained in:
duanshuwen
717
tests/test_api_contracts.py
Normal file
717
tests/test_api_contracts.py
Normal file
@@ -0,0 +1,717 @@
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app import serializers
|
||||
from app.auth import hash_password, require_admin
|
||||
from app.database import get_db
|
||||
from app.main import create_app
|
||||
from app.models import AdminUser, CtaBanner, Destination, DestinationAlias, HeroSlide, Lead, MediaAsset, Product, ProductImage, ThemeCard
|
||||
from app.routers import admin as admin_router
|
||||
from app.routers.admin import normalize_detail_sections, normalize_images
|
||||
from app.schemas import (
|
||||
AdminProductQuery,
|
||||
LeadCreateIn,
|
||||
LeadQuery,
|
||||
ProductCreateIn,
|
||||
ProductImageIn,
|
||||
ProductQuery,
|
||||
)
|
||||
|
||||
|
||||
class FakeScalarResult:
|
||||
def __init__(self, items):
|
||||
self.items = items
|
||||
|
||||
def unique(self):
|
||||
return self
|
||||
|
||||
def all(self):
|
||||
return self.items
|
||||
|
||||
def first(self):
|
||||
return self.items[0] if self.items else None
|
||||
|
||||
def one(self):
|
||||
return self.items[0]
|
||||
|
||||
|
||||
class FakeExecuteResult:
|
||||
def __init__(self, rows):
|
||||
self.rows = rows
|
||||
|
||||
def all(self):
|
||||
return self.rows
|
||||
|
||||
|
||||
class FakeDb:
|
||||
def __init__(self, *, scalar_results=None, scalar_values=None, execute_results=None, get_result=None):
|
||||
self.scalar_results = list(scalar_results or [])
|
||||
self.scalar_values = list(scalar_values or [])
|
||||
self.execute_results = list(execute_results or [])
|
||||
self.get_result = get_result
|
||||
self.added = []
|
||||
self.deleted = []
|
||||
self.committed = False
|
||||
|
||||
def scalars(self, _stmt):
|
||||
return FakeScalarResult(self.scalar_results.pop(0))
|
||||
|
||||
def scalar(self, _stmt):
|
||||
return self.scalar_values.pop(0)
|
||||
|
||||
def execute(self, _stmt):
|
||||
return FakeExecuteResult(self.execute_results.pop(0))
|
||||
|
||||
def get(self, _model, _item_id):
|
||||
return 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 refresh(self, item):
|
||||
if not getattr(item, "id", None):
|
||||
item.id = "lead-test-id"
|
||||
|
||||
|
||||
def make_destination(**overrides):
|
||||
destination = Destination(
|
||||
id=overrides.get("id", "dest-test"),
|
||||
name=overrides.get("name", "测试目的地"),
|
||||
slug=overrides.get("slug", "test-destination"),
|
||||
region=overrides.get("region", "测试区域"),
|
||||
image=overrides.get("image", "/assets/test.jpg"),
|
||||
isHot=overrides.get("isHot", True),
|
||||
sortOrder=overrides.get("sortOrder", 1),
|
||||
isActive=overrides.get("isActive", True),
|
||||
createdAt=datetime(2026, 1, 1),
|
||||
updatedAt=datetime(2026, 1, 2),
|
||||
)
|
||||
destination.aliases = [
|
||||
DestinationAlias(id="alias-b", alias="别名B", destinationId=destination.id),
|
||||
DestinationAlias(id="alias-a", alias="别名A", destinationId=destination.id),
|
||||
]
|
||||
destination.products = []
|
||||
return destination
|
||||
|
||||
|
||||
def make_product(**overrides):
|
||||
destination = overrides.get("destination", make_destination())
|
||||
product = Product(
|
||||
id=overrides.get("id", "product-test"),
|
||||
sourceId=overrides.get("sourceId", 101),
|
||||
title=overrides.get("title", "测试线路"),
|
||||
subtitle=overrides.get("subtitle", "测试副标题"),
|
||||
destinationId=destination.id,
|
||||
priceAmount=overrides.get("priceAmount", 1000),
|
||||
priceUnit=overrides.get("priceUnit", "起/人"),
|
||||
tags=overrides.get("tags", ["测试", "线路"]),
|
||||
coverImage=overrides.get("coverImage", "/assets/cover.jpg"),
|
||||
summary=overrides.get("summary", "测试摘要"),
|
||||
detailSections=overrides.get("detailSections", None),
|
||||
status=overrides.get("status", "published"),
|
||||
sortWeight=overrides.get("sortWeight", 1),
|
||||
publishedAt=datetime(2026, 1, 3),
|
||||
createdAt=datetime(2026, 1, 1),
|
||||
updatedAt=datetime(2026, 1, 2),
|
||||
)
|
||||
product.destination = destination
|
||||
product.images = [
|
||||
ProductImage(id="img-b", productId=product.id, url="/assets/b.jpg", alt="B", sortOrder=2),
|
||||
ProductImage(id="img-a", productId=product.id, url="/assets/a.jpg", alt="A", sortOrder=1),
|
||||
]
|
||||
return product
|
||||
|
||||
|
||||
def make_hero_slide(**overrides):
|
||||
return HeroSlide(
|
||||
id=overrides.get("id", "slide-test"),
|
||||
title=overrides.get("title", "测试轮播"),
|
||||
kicker=overrides.get("kicker", "测试副标题"),
|
||||
image=overrides.get("image", "/assets/slide.jpg"),
|
||||
targetType=overrides.get("targetType", "campaign"),
|
||||
targetValue=overrides.get("targetValue", "campaign-test"),
|
||||
sortOrder=overrides.get("sortOrder", 1),
|
||||
isActive=overrides.get("isActive", True),
|
||||
createdAt=datetime(2026, 1, 1),
|
||||
updatedAt=datetime(2026, 1, 2),
|
||||
)
|
||||
|
||||
|
||||
def make_theme_card(**overrides):
|
||||
return ThemeCard(
|
||||
id=overrides.get("id", "theme-test"),
|
||||
label=overrides.get("label", "测试主题"),
|
||||
image=overrides.get("image", "/assets/theme.jpg"),
|
||||
targetType=overrides.get("targetType", "search"),
|
||||
targetValue=overrides.get("targetValue", "测试主题"),
|
||||
sortOrder=overrides.get("sortOrder", 1),
|
||||
isActive=overrides.get("isActive", True),
|
||||
createdAt=datetime(2026, 1, 1),
|
||||
updatedAt=datetime(2026, 1, 2),
|
||||
)
|
||||
|
||||
|
||||
def make_cta_banner(**overrides):
|
||||
return CtaBanner(
|
||||
id=overrides.get("id", "cta-test"),
|
||||
alt=overrides.get("alt", "测试运营入口"),
|
||||
image=overrides.get("image", "/assets/cta.jpg"),
|
||||
targetType=overrides.get("targetType", "lead"),
|
||||
targetValue=overrides.get("targetValue", "cta-test"),
|
||||
sortOrder=overrides.get("sortOrder", 1),
|
||||
isActive=overrides.get("isActive", True),
|
||||
createdAt=datetime(2026, 1, 1),
|
||||
updatedAt=datetime(2026, 1, 2),
|
||||
)
|
||||
|
||||
|
||||
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 test_public_product_serializer_limits_destination_and_stabilizes_arrays():
|
||||
data = serializers.public_product_dict(make_product())
|
||||
|
||||
assert data["destination"] == {"id": "dest-test", "name": "测试目的地"}
|
||||
assert [image["id"] for image in data["images"]] == ["img-a", "img-b"]
|
||||
assert data["detailSections"] == []
|
||||
|
||||
|
||||
def test_destination_serializer_uses_precomputed_product_count_and_sorts_aliases():
|
||||
destination = make_destination()
|
||||
|
||||
data = serializers.destination_dict(destination, include_count=True, product_count=7)
|
||||
|
||||
assert data["_count"] == {"products": 7}
|
||||
assert [alias["id"] for alias in data["aliases"]] == ["alias-a", "alias-b"]
|
||||
|
||||
|
||||
def test_lead_schema_accepts_date_only_string():
|
||||
lead = LeadCreateIn(phone=" contact handle ", travelDate="2027-01-01")
|
||||
|
||||
assert lead.phone == "contact handle"
|
||||
assert lead.travelDate == datetime(2027, 1, 1)
|
||||
|
||||
|
||||
def test_query_status_validation_rejects_unknown_values():
|
||||
with pytest.raises(ValidationError):
|
||||
ProductQuery(status="hidden")
|
||||
with pytest.raises(ValidationError):
|
||||
AdminProductQuery(status="hidden")
|
||||
with pytest.raises(ValidationError):
|
||||
LeadQuery(status="closed")
|
||||
|
||||
|
||||
def test_normalize_images_reassigns_sort_order_by_payload_order():
|
||||
images = [
|
||||
ProductImageIn(url="/assets/first.jpg", sortOrder=99),
|
||||
ProductImageIn(url="/assets/second.jpg", sortOrder=10),
|
||||
]
|
||||
|
||||
assert [image["sortOrder"] for image in normalize_images(images)] == [0, 1]
|
||||
|
||||
|
||||
def test_empty_detail_sections_are_accepted_and_dropped():
|
||||
body = ProductCreateIn(
|
||||
title="测试线路",
|
||||
detailSections=[{"key": "", "label": "", "blocks": []}],
|
||||
)
|
||||
|
||||
assert normalize_detail_sections(body.detailSections) == []
|
||||
|
||||
|
||||
def test_public_products_endpoint_uses_public_product_contract():
|
||||
product = make_product()
|
||||
fake_db = FakeDb(scalar_results=[[product]])
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: fake_db
|
||||
|
||||
try:
|
||||
response = TestClient(app).get("/api/public/products")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
item = response.json()["items"][0]
|
||||
assert item["destination"] == {"id": "dest-test", "name": "测试目的地"}
|
||||
assert item["images"][0]["url"] == "/assets/a.jpg"
|
||||
|
||||
|
||||
def test_public_leads_endpoint_accepts_date_only_and_returns_minimal_response():
|
||||
fake_db = FakeDb()
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: fake_db
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(
|
||||
"/api/public/leads",
|
||||
json={
|
||||
"phone": " contact handle ",
|
||||
"travelDate": "2027-01-01",
|
||||
"peopleCount": 2,
|
||||
"budgetMin": 0,
|
||||
"budgetMax": 100,
|
||||
},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.json() == {"id": "lead-test-id", "status": "new"}
|
||||
assert fake_db.added[0].phone == "contact handle"
|
||||
assert fake_db.added[0].travelDate == datetime(2027, 1, 1)
|
||||
|
||||
|
||||
def test_admin_destinations_endpoint_uses_precomputed_product_counts():
|
||||
destination = make_destination()
|
||||
fake_db = FakeDb(scalar_results=[[destination]], execute_results=[[(destination.id, 7)]])
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: fake_db
|
||||
app.dependency_overrides[require_admin] = lambda: AdminUser(
|
||||
id="admin-test",
|
||||
email="admin@example.test",
|
||||
name="Admin",
|
||||
role="admin",
|
||||
passwordHash="not-used",
|
||||
isActive=True,
|
||||
)
|
||||
|
||||
try:
|
||||
response = TestClient(app).get("/api/admin/destinations")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["items"][0]["_count"] == {"products": 7}
|
||||
|
||||
|
||||
def test_admin_requires_auth_for_protected_endpoint():
|
||||
response = TestClient(create_app()).get("/api/admin/products")
|
||||
|
||||
assert response.status_code == 401
|
||||
assert response.json() == {"message": "请先登录后台"}
|
||||
|
||||
|
||||
def test_admin_login_returns_token_and_user():
|
||||
admin_user = AdminUser(
|
||||
id="admin-test",
|
||||
email="admin@example.com",
|
||||
name="Admin",
|
||||
role="admin",
|
||||
passwordHash=hash_password("ChangeMe123!", rounds=4),
|
||||
isActive=True,
|
||||
)
|
||||
fake_db = FakeDb(scalar_values=[admin_user])
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: fake_db
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(
|
||||
"/api/admin/auth/login",
|
||||
json={"email": "admin@example.com", "password": "ChangeMe123!"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["token"]
|
||||
assert body["user"] == {
|
||||
"id": "admin-test",
|
||||
"email": "admin@example.com",
|
||||
"name": "Admin",
|
||||
"role": "admin",
|
||||
}
|
||||
|
||||
|
||||
def test_site_config_patch_ignores_fields_not_allowed_for_module_and_audits():
|
||||
destination = make_destination()
|
||||
fake_db = FakeDb(get_result=destination)
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: fake_db
|
||||
app.dependency_overrides[require_admin] = lambda: AdminUser(
|
||||
id="admin-test",
|
||||
email="admin@example.test",
|
||||
name="Admin",
|
||||
role="admin",
|
||||
passwordHash="not-used",
|
||||
isActive=True,
|
||||
)
|
||||
|
||||
try:
|
||||
response = TestClient(app).patch(
|
||||
"/api/admin/site-config/destinations/dest-test",
|
||||
json={"name": "新目的地", "targetType": "ignored"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == "新目的地"
|
||||
assert not hasattr(destination, "targetType")
|
||||
assert fake_db.committed
|
||||
assert fake_db.added[-1].entity == "destination"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("module", "payload", "expected"),
|
||||
[
|
||||
("heroSlides", {"title": "新轮播", "image": None, "targetType": None}, {"title": "新轮播", "image": None}),
|
||||
("destinations", {"name": "新目的地", "slug": "", "region": None, "isHot": True}, {"name": "新目的地", "slug": "e696b0e79baee79a84e59cb0", "isHot": True}),
|
||||
("themes", {"label": "新主题", "image": None}, {"label": "新主题", "image": ""}),
|
||||
("ctaBanners", {"alt": "新运营入口", "image": None, "targetType": None}, {"alt": "新运营入口", "image": "", "targetType": ""}),
|
||||
],
|
||||
)
|
||||
def test_site_config_create_modules_defaults_fields_and_audits(module, payload, expected):
|
||||
fake_db = FakeDb(scalar_values=[4])
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(f"/api/admin/site-config/{module}", json=payload)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 201
|
||||
body = response.json()
|
||||
for key, value in expected.items():
|
||||
assert body[key] == value
|
||||
assert body["isActive"] is True
|
||||
assert body["sortOrder"] == 5
|
||||
assert fake_db.committed
|
||||
assert fake_db.added[-1].action == "create"
|
||||
|
||||
|
||||
def test_site_config_invalid_module_returns_structured_error():
|
||||
fake_db = FakeDb()
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).post("/api/admin/site-config/unknown", json={"title": "测试"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json() == {
|
||||
"message": "模块不存在或无权限操作",
|
||||
"code": "MODULE_CONFIG_FORBIDDEN",
|
||||
"details": {"module": "unknown"},
|
||||
}
|
||||
|
||||
|
||||
def test_site_config_create_requires_module_primary_field():
|
||||
fake_db = FakeDb()
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).post("/api/admin/site-config/themes", json={"image": "/assets/theme.jpg"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 422
|
||||
assert response.json()["code"] == "MODULE_CONFIG_VALIDATION_ERROR"
|
||||
assert response.json()["details"] == {"field": "label"}
|
||||
|
||||
|
||||
def test_admin_site_config_hero_slides_use_dedicated_contract_without_targets():
|
||||
hero_slide = make_hero_slide(targetType="campaign", targetValue="campaign-test")
|
||||
fake_db = FakeDb(scalar_results=[[], [hero_slide], [], []])
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).get("/api/admin/site-config")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
hero = response.json()["heroSlides"][0]
|
||||
assert hero["title"] == "测试轮播"
|
||||
assert hero["image"] == "/assets/slide.jpg"
|
||||
assert "targetType" not in hero
|
||||
assert "targetValue" not in hero
|
||||
|
||||
|
||||
def test_site_config_create_hero_slide_ignores_target_fields_and_returns_dedicated_contract():
|
||||
fake_db = FakeDb(scalar_values=[0])
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(
|
||||
"/api/admin/site-config/heroSlides",
|
||||
json={
|
||||
"title": " 新轮播 ",
|
||||
"kicker": "",
|
||||
"image": None,
|
||||
"targetType": "campaign",
|
||||
"targetValue": "ignored",
|
||||
},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 201
|
||||
body = response.json()
|
||||
assert body["title"] == "新轮播"
|
||||
assert body["image"] is None
|
||||
assert "targetType" not in body
|
||||
assert "targetValue" not in body
|
||||
created = fake_db.added[0]
|
||||
assert created.targetType is None
|
||||
assert created.targetValue is None
|
||||
|
||||
|
||||
def test_site_config_patch_hero_slide_ignores_target_fields_and_returns_dedicated_contract():
|
||||
hero_slide = make_hero_slide(targetType="campaign", targetValue="campaign-test")
|
||||
fake_db = FakeDb(get_result=hero_slide)
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).patch(
|
||||
"/api/admin/site-config/heroSlides/slide-test",
|
||||
json={
|
||||
"title": "夏日贵州小包团",
|
||||
"image": None,
|
||||
"targetType": "search",
|
||||
"targetValue": "ignored",
|
||||
},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["title"] == "夏日贵州小包团"
|
||||
assert body["image"] is None
|
||||
assert "targetType" not in body
|
||||
assert "targetValue" not in body
|
||||
assert hero_slide.targetType == "campaign"
|
||||
assert hero_slide.targetValue == "campaign-test"
|
||||
|
||||
|
||||
def test_site_config_patch_updates_destination_contract_fields():
|
||||
destination = make_destination()
|
||||
fake_db = FakeDb(get_result=destination)
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).patch(
|
||||
"/api/admin/site-config/destinations/dest-test",
|
||||
json={"slug": "new-slug", "region": None, "image": None, "isHot": False, "sortOrder": 8, "targetType": "ignored"},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["slug"] == "new-slug"
|
||||
assert body["region"] is None
|
||||
assert body["image"] is None
|
||||
assert body["isHot"] is False
|
||||
assert body["sortOrder"] == 8
|
||||
assert not hasattr(destination, "targetType")
|
||||
assert fake_db.added[-1].action == "update"
|
||||
|
||||
|
||||
def test_site_config_delete_returns_json_reorders_remaining_items_and_audits():
|
||||
delete_item = make_theme_card(id="theme-delete", sortOrder=1)
|
||||
remaining = [make_theme_card(id="theme-b", sortOrder=5), make_theme_card(id="theme-a", sortOrder=9)]
|
||||
fake_db = FakeDb(get_result=delete_item, scalar_results=[remaining])
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).delete("/api/admin/site-config/themes/theme-delete")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"id": "theme-delete"}
|
||||
assert fake_db.deleted == [delete_item]
|
||||
assert [item.sortOrder for item in remaining] == [0, 1]
|
||||
assert fake_db.committed
|
||||
assert fake_db.added[-1].action == "delete"
|
||||
assert fake_db.added[-1].entity == "theme_card"
|
||||
|
||||
|
||||
def test_site_config_delete_destination_with_products_returns_conflict():
|
||||
destination = make_destination()
|
||||
fake_db = FakeDb(get_result=destination, scalar_values=[1])
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).delete("/api/admin/site-config/destinations/dest-test")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 409
|
||||
assert response.json()["code"] == "MODULE_CONFIG_CONFLICT"
|
||||
assert not fake_db.deleted
|
||||
|
||||
|
||||
def test_site_config_reorder_reassigns_sort_order_and_returns_items():
|
||||
items = [
|
||||
make_hero_slide(id="slide-1", title="第一张", sortOrder=0),
|
||||
make_hero_slide(id="slide-2", title="第二张", sortOrder=1),
|
||||
make_hero_slide(id="slide-3", title="第三张", sortOrder=2),
|
||||
]
|
||||
fake_db = FakeDb(scalar_results=[items])
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).patch(
|
||||
"/api/admin/site-config/heroSlides/reorder",
|
||||
json={"itemIds": ["slide-2", "slide-1", "slide-3"]},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert [item.id for item in items] == ["slide-1", "slide-2", "slide-3"]
|
||||
assert {item.id: item.sortOrder for item in items} == {"slide-2": 0, "slide-1": 1, "slide-3": 2}
|
||||
assert [item["id"] for item in response.json()["items"]] == ["slide-2", "slide-1", "slide-3"]
|
||||
assert [item["sortOrder"] for item in response.json()["items"]] == [0, 1, 2]
|
||||
assert fake_db.committed
|
||||
assert fake_db.added[-1].action == "reorder"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"item_ids",
|
||||
[
|
||||
["slide-1", "slide-1", "slide-2"],
|
||||
["slide-1", "slide-2"],
|
||||
["slide-1", "slide-2", "other-module-id"],
|
||||
],
|
||||
)
|
||||
def test_site_config_reorder_rejects_duplicate_missing_and_unknown_ids(item_ids):
|
||||
items = [
|
||||
make_hero_slide(id="slide-1"),
|
||||
make_hero_slide(id="slide-2"),
|
||||
make_hero_slide(id="slide-3"),
|
||||
]
|
||||
fake_db = FakeDb(scalar_results=[items])
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).patch("/api/admin/site-config/heroSlides/reorder", json={"itemIds": item_ids})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json()["code"] == "MODULE_CONFIG_REORDER_INVALID"
|
||||
assert not fake_db.committed
|
||||
|
||||
|
||||
def test_admin_media_upload_streams_image_to_oss_records_asset_and_audits(monkeypatch):
|
||||
uploaded = {}
|
||||
|
||||
def fake_upload(file_obj, key, mime_type, size_bytes):
|
||||
uploaded["key"] = key
|
||||
uploaded["mimeType"] = mime_type
|
||||
uploaded["sizeBytes"] = size_bytes
|
||||
uploaded["body"] = file_obj.read()
|
||||
return f"https://cdn.example.test/{key}"
|
||||
|
||||
monkeypatch.setattr(admin_router, "upload_image_to_oss", fake_upload)
|
||||
fake_db = FakeDb()
|
||||
app = authenticated_app(fake_db)
|
||||
content = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR"
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(
|
||||
"/api/admin/media-assets/upload",
|
||||
data={"group": "heroSlides"},
|
||||
files={"file": ("hero.png", content, "image/png")},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 201
|
||||
body = response.json()
|
||||
assert body["url"].startswith("https://cdn.example.test/admin/heroSlides/")
|
||||
assert body["name"] == "hero.png"
|
||||
assert body["mimeType"] == "image/png"
|
||||
assert body["sizeBytes"] == len(content)
|
||||
assert body["group"] == "heroSlides"
|
||||
assert uploaded["body"] == content
|
||||
assert uploaded["key"].endswith(".png")
|
||||
assert isinstance(fake_db.added[0], MediaAsset)
|
||||
assert fake_db.added[-1].entity == "media_asset"
|
||||
assert fake_db.committed
|
||||
|
||||
|
||||
def test_admin_media_upload_rejects_non_image_file():
|
||||
fake_db = FakeDb()
|
||||
app = authenticated_app(fake_db)
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(
|
||||
"/api/admin/media-assets/upload",
|
||||
data={"group": "docs"},
|
||||
files={"file": ("note.txt", b"hello", "text/plain")},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json()["code"] == "MEDIA_UPLOAD_INVALID_TYPE"
|
||||
assert not fake_db.added
|
||||
assert not fake_db.committed
|
||||
|
||||
|
||||
def test_admin_lead_status_update_returns_updated_status_and_audits():
|
||||
lead = Lead(
|
||||
id="lead-test",
|
||||
destination="测试目的地",
|
||||
phone="contact handle",
|
||||
status="new",
|
||||
createdAt=datetime(2026, 1, 1),
|
||||
updatedAt=datetime(2026, 1, 2),
|
||||
)
|
||||
fake_db = FakeDb(get_result=lead)
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: fake_db
|
||||
app.dependency_overrides[require_admin] = lambda: AdminUser(
|
||||
id="admin-test",
|
||||
email="admin@example.test",
|
||||
name="Admin",
|
||||
role="admin",
|
||||
passwordHash="not-used",
|
||||
isActive=True,
|
||||
)
|
||||
|
||||
try:
|
||||
response = TestClient(app).patch("/api/admin/leads/lead-test/status", json={"status": "contacted"})
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["status"] == "contacted"
|
||||
assert lead.status == "contacted"
|
||||
assert fake_db.committed
|
||||
assert fake_db.added[-1].entity == "lead"
|
||||
Reference in New Issue
Block a user