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, Campaign, CtaBanner, Destination, DestinationAlias, DestinationHero, DestinationRegion, HeroSlide, HotelGroup, Lead, MapImage, MediaAsset, Product, ProductImage, RouteSection, RouteSectionProduct, ThemeCard, VehicleOption from app.routers import admin as admin_router from app.routers.admin import normalize_detail_sections, normalize_images from app.routers.shared import site_config 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_hotel_group(**overrides): return HotelGroup( id=overrides.get("id", "hotel-group-test"), title=overrides.get("title", "经典酒店"), description=overrides.get("description", "经典城市酒店与山野度假住宿组合"), image=overrides.get("image", "/assets/hotel.jpg"), coverImage=overrides.get("coverImage", overrides.get("image", "/assets/hotel.jpg")), priceAmount=overrides.get("priceAmount", 68000), priceUnit=overrides.get("priceUnit", "起/晚"), tags=overrides.get("tags", ["温泉", "亲子"]), status=overrides.get("status", "published"), sortOrder=overrides.get("sortOrder", 1), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_vehicle_option(**overrides): return VehicleOption( id=overrides.get("id", "vehicle-option-test"), title=overrides.get("title", "5座舒适用车"), description=overrides.get("description", "2-8人小团，按人数和行李匹配车型"), image=overrides.get("image", "/assets/vehicle.jpg"), sortOrder=overrides.get("sortOrder", 2), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_map_image(**overrides): return MapImage( id=overrides.get("id", "map-test"), image=overrides.get("image", "https://cdn.example.test/map.webp"), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_destination_hero(**overrides): return DestinationHero( id=overrides.get("id", "destination-hero-test"), title=overrides.get("title", "目的地页主视觉"), kicker=overrides.get("kicker", "贵州小包团目的地"), image=overrides.get("image", "/assets/destination-hero.jpg"), sortOrder=overrides.get("sortOrder", 0), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_destination_region(**overrides): return DestinationRegion( id=overrides.get("id", "destination-region-test"), label=overrides.get("label", "黔南"), keyword=overrides.get("keyword", "黔南荔波"), spots=overrides.get("spots", "荔波小七孔、茂兰"), sortOrder=overrides.get("sortOrder", 0), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_campaign(**overrides): return Campaign( id=overrides.get("id", "campaign-test"), slug=overrides.get("slug", "classic-deal"), title=overrides.get("title", "经典打卡特惠"), description=overrides.get("description", "经典首游活动"), coverImage=overrides.get("coverImage", "https://cdn.example.test/campaign.webp"), priceAmount=overrides.get("priceAmount", 162500), priceUnit=overrides.get("priceUnit", "起/人"), tags=overrides.get("tags", ["臻藏旅位", "赛事庆典"]), status=overrides.get("status", "draft"), startsAt=overrides.get("startsAt"), endsAt=overrides.get("endsAt"), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_route_section(**overrides): section = RouteSection( id=overrides.get("id", "routes"), title=overrides.get("title", "经典人文打卡线路"), subtitle=overrides.get("subtitle", "黄果树、荔波小七孔、千户苗寨、镇远古城、梵净山一次串联"), sortOrder=overrides.get("sortOrder", 0), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) section.products = overrides.get("products", []) return section def make_route_section_product(**overrides): product = overrides.get("product", make_product(id=overrides.get("productId", "product-test"))) link = RouteSectionProduct( sectionId=overrides.get("sectionId", "routes"), productId=overrides.get("productId", product.id), sortOrder=overrides.get("sortOrder", 0), ) link.product = product return link 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": ""}), ("destinationHero", {"title": "目的地页主视觉", "kicker": "贵州小包团目的地", "image": None}, {"title": "目的地页主视觉", "kicker": "贵州小包团目的地", "image": None}), ("destinationRegions", {"label": "黔南", "keyword": "黔南荔波", "spots": "荔波小七孔、茂兰"}, {"label": "黔南", "keyword": "黔南荔波", "spots": "荔波小七孔、茂兰"}), ("ctaBanners", {"alt": "新运营入口", "image": None, "targetType": None}, {"alt": "新运营入口", "image": "", "targetType": ""}), ("hotelGroups", {"title": "经典酒店", "description": "酒店文案", "image": None}, {"title": "经典酒店", "description": "酒店文案", "image": None}), ("vehicleOptions", {"title": "5座舒适用车", "description": "用车文案", "image": None}, {"title": "5座舒适用车", "description": "用车文案", "image": None}), ], ) 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_admin_site_config_includes_map_array_with_dedicated_contract(): map_image = make_map_image() fake_db = FakeDb(scalar_results=[[], [], [], [], [map_image], [], [], [], [], [], []]) app = authenticated_app(fake_db) try: response = TestClient(app).get("/api/admin/site-config") finally: app.dependency_overrides.clear() assert response.status_code == 200 body = response.json() assert body["map"] == [ { "id": "map-test", "image": "https://cdn.example.test/map.webp", "isActive": True, "createdAt": "2026-01-01T00:00:00", "updatedAt": "2026-01-02T00:00:00", } ] assert "sortOrder" not in body["map"][0] assert "title" not in body["map"][0] assert "targetType" not in body["map"][0] def test_admin_site_config_includes_destination_page_modules(): fake_db = FakeDb(scalar_results=[[], [], [], [], [], [], [], [], [], [], []]) app = authenticated_app(fake_db) try: response = TestClient(app).get("/api/admin/site-config") finally: app.dependency_overrides.clear() assert response.status_code == 200 body = response.json() assert body["destinationHero"] == [] assert body["destinationRegions"] == [] def test_public_site_config_includes_destination_page_modules(): destination_hero = make_destination_hero() destination_region = make_destination_region() fake_db = FakeDb(scalar_results=[[], [], [destination_hero], [destination_region], [], [], [], [], [], [], []]) result = site_config(fake_db, active_only=True) assert result["destinationHero"] == [ { "id": "destination-hero-test", "title": "目的地页主视觉", "kicker": "贵州小包团目的地", "image": "/assets/destination-hero.jpg", "sortOrder": 0, "isActive": True, "createdAt": "2026-01-01T00:00:00", "updatedAt": "2026-01-02T00:00:00", } ] assert result["destinationRegions"] == [ { "id": "destination-region-test", "label": "黔南", "keyword": "黔南荔波", "spots": "荔波小七孔、茂兰", "sortOrder": 0, "isActive": True, "createdAt": "2026-01-01T00:00:00", "updatedAt": "2026-01-02T00:00:00", } ] def test_admin_site_config_includes_campaigns_for_special_offers(): campaign = make_campaign(status="published") fake_db = FakeDb(scalar_results=[[], [], [], [], [], [], [], [campaign], [], [], []]) app = authenticated_app(fake_db) try: response = TestClient(app).get("/api/admin/site-config") finally: app.dependency_overrides.clear() assert response.status_code == 200 body = response.json() assert body["campaigns"] == [ { "id": "campaign-test", "slug": "classic-deal", "title": "经典打卡特惠", "description": "经典首游活动", "coverImage": "https://cdn.example.test/campaign.webp", "priceAmount": 162500, "priceUnit": "起/人", "tags": ["臻藏旅位", "赛事庆典"], "status": "published", "startsAt": None, "endsAt": None, "createdAt": "2026-01-01T00:00:00", "updatedAt": "2026-01-02T00:00:00", } ] assert "sortOrder" not in body["campaigns"][0] assert "isActive" not in body["campaigns"][0] assert "targetType" not in body["campaigns"][0] def test_admin_site_config_includes_route_sections_for_featured_routes(): section = make_route_section( products=[ make_route_section_product(productId="product-a", sortOrder=1), make_route_section_product(productId="product-b", sortOrder=0), ], ) fake_db = FakeDb(scalar_results=[[], [], [], [], [], [], [], [], [section], [], []]) app = authenticated_app(fake_db) try: response = TestClient(app).get("/api/admin/site-config") finally: app.dependency_overrides.clear() assert response.status_code == 200 route_sections = response.json()["routeSections"] assert route_sections == [ { "id": "routes", "title": "经典人文打卡线路", "subtitle": "黄果树、荔波小七孔、千户苗寨、镇远古城、梵净山一次串联", "productIds": ["product-b", "product-a"], "isActive": True, "sortOrder": 0, "createdAt": "2026-01-01T00:00:00", "updatedAt": "2026-01-02T00:00:00", } ] def test_admin_site_config_includes_hotel_and_vehicle_modules(): hotel_group = make_hotel_group() vehicle_option = make_vehicle_option() fake_db = FakeDb(scalar_results=[[], [], [], [], [], [], [], [], [], [hotel_group], [vehicle_option]]) app = authenticated_app(fake_db) try: response = TestClient(app).get("/api/admin/site-config") finally: app.dependency_overrides.clear() assert response.status_code == 200 body = response.json() assert body["hotelGroups"] == [ { "id": "hotel-group-test", "title": "经典酒店", "description": "经典城市酒店与山野度假住宿组合", "image": "/assets/hotel.jpg", "coverImage": "/assets/hotel.jpg", "priceAmount": 68000, "priceUnit": "起/晚", "tags": ["温泉", "亲子"], "status": "published", "sortOrder": 1, "isActive": True, "createdAt": "2026-01-01T00:00:00", "updatedAt": "2026-01-02T00:00:00", } ] assert body["vehicleOptions"] == [ { "id": "vehicle-option-test", "title": "5座舒适用车", "description": "2-8人小团，按人数和行李匹配车型", "image": "/assets/vehicle.jpg", "sortOrder": 2, "isActive": True, "createdAt": "2026-01-01T00:00:00", "updatedAt": "2026-01-02T00:00:00", } ] def test_site_config_create_route_section_generates_dynamic_id_and_audits(): fake_db = FakeDb(scalar_results=[[]]) app = authenticated_app(fake_db) try: response = TestClient(app).post( "/api/admin/site-config/routeSections", json={"title": " 新人文线路 ", "subtitle": " 新副文案 "}, ) finally: app.dependency_overrides.clear() assert response.status_code == 201 body = response.json() assert body["id"].startswith("routesection-") assert body["title"] == "新人文线路" assert body["subtitle"] == "新副文案" assert body["productIds"] == [] assert body["isActive"] is True assert body["sortOrder"] == 0 created = fake_db.added[0] assert isinstance(created, RouteSection) assert created.id.startswith("routesection-") assert fake_db.added[-1].entity == "route_section" assert fake_db.committed def test_route_sections_module_is_not_fixed(): assert admin_router.SITE_CONFIG_MODULES["routeSections"].get("fixed") is not True def test_site_config_create_route_section_allows_more_than_three_groups(): items = [ make_route_section(id="route-section-a", title="经典", sortOrder=0), make_route_section(id="route-section-b", title="户外", sortOrder=1), make_route_section(id="route-section-c", title="混搭", sortOrder=2), ] fake_db = FakeDb(scalar_results=[items]) app = authenticated_app(fake_db) try: response = TestClient(app).post("/api/admin/site-config/routeSections", json={"title": "第四组"}) finally: app.dependency_overrides.clear() assert response.status_code == 201 body = response.json() assert body["title"] == "第四组" assert body["sortOrder"] == 3 assert body["productIds"] == [] assert fake_db.committed 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_create_map_image_defaults_active_and_audits(): fake_db = FakeDb(scalar_results=[[]]) app = authenticated_app(fake_db) try: response = TestClient(app).post( "/api/admin/site-config/map", json={"image": " https://cdn.example.test/guizhou-map.webp ", "title": "ignored", "sortOrder": 99}, ) finally: app.dependency_overrides.clear() assert response.status_code == 201 body = response.json() assert body["image"] == "https://cdn.example.test/guizhou-map.webp" assert body["isActive"] is True assert "title" not in body assert "sortOrder" not in body created = fake_db.added[0] assert isinstance(created, MapImage) assert created.image == "https://cdn.example.test/guizhou-map.webp" assert created.isActive is True assert fake_db.added[-1].entity == "map_image" assert fake_db.committed def test_site_config_create_map_image_rejects_duplicate_singleton(): fake_db = FakeDb(scalar_results=[[make_map_image()]]) app = authenticated_app(fake_db) try: response = TestClient(app).post("/api/admin/site-config/map", json={"image": "https://cdn.example.test/new.webp"}) finally: app.dependency_overrides.clear() assert response.status_code == 409 assert response.json()["code"] == "MAP_IMAGE_ALREADY_EXISTS" assert not fake_db.added assert not fake_db.committed def test_site_config_create_campaign_defaults_draft_and_audits(): fake_db = FakeDb() app = authenticated_app(fake_db) try: response = TestClient(app).post( "/api/admin/site-config/campaigns", json={ "title": " Classic Deal ", "slug": " classic-deal ", "description": " Deal summary ", "coverImage": " https://cdn.example.test/campaign.webp ", "priceAmount": 188000, "priceUnit": " 起/人 ", "tags": [" 季节限定 ", "", "小众秘境"], "image": "ignored", "isActive": True, "sortOrder": 9, }, ) finally: app.dependency_overrides.clear() assert response.status_code == 201 body = response.json() assert body["title"] == "Classic Deal" assert body["slug"] == "classic-deal" assert body["description"] == "Deal summary" assert body["coverImage"] == "https://cdn.example.test/campaign.webp" assert body["priceAmount"] == 188000 assert body["priceUnit"] == "起/人" assert body["tags"] == ["季节限定", "小众秘境"] assert body["status"] == "draft" assert "image" not in body assert "isActive" not in body assert "sortOrder" not in body created = fake_db.added[0] assert isinstance(created, Campaign) assert created.status == "draft" assert created.priceAmount == 188000 assert created.priceUnit == "起/人" assert created.tags == ["季节限定", "小众秘境"] assert fake_db.added[-1].entity == "campaign" assert fake_db.committed def test_site_config_create_campaign_rejects_more_than_three_tags(): fake_db = FakeDb() app = authenticated_app(fake_db) try: response = TestClient(app).post( "/api/admin/site-config/campaigns", json={ "title": "Classic Deal", "slug": "classic-deal", "tags": ["a", "b", "c", "d"], }, ) finally: app.dependency_overrides.clear() assert response.status_code == 422 assert response.json()["code"] == "MODULE_CONFIG_VALIDATION_ERROR" assert response.json()["details"] == {"field": "tags", "max": 3} assert not fake_db.added assert not fake_db.committed 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_map_image_updates_allowed_fields_only(): map_image = make_map_image(isActive=True) fake_db = FakeDb(get_result=map_image) app = authenticated_app(fake_db) try: response = TestClient(app).patch( "/api/admin/site-config/map/map-test", json={"image": " https://cdn.example.test/new-map.webp ", "isActive": False, "targetType": "ignored", "sortOrder": 1}, ) finally: app.dependency_overrides.clear() assert response.status_code == 200 body = response.json() assert body["image"] == "https://cdn.example.test/new-map.webp" assert body["isActive"] is False assert "targetType" not in body assert "sortOrder" not in body assert not hasattr(map_image, "targetType") assert fake_db.added[-1].action == "update" assert fake_db.added[-1].entity == "map_image" def test_site_config_patch_campaign_updates_contract_fields_only(): campaign = make_campaign(status="draft") fake_db = FakeDb(get_result=campaign) app = authenticated_app(fake_db) try: response = TestClient(app).patch( "/api/admin/site-config/campaigns/campaign-test", json={ "title": " Summer Deal ", "slug": " summer-deal ", "description": None, "coverImage": " https://cdn.example.test/summer.webp ", "priceAmount": 57500, "priceUnit": " 起/人 ", "tags": ["无可比拟", "尊享礼遇", ""], "status": "published", "startsAt": "2026-07-01T08:00:00", "endsAt": None, "image": "ignored", "isActive": False, "sortOrder": 3, }, ) finally: app.dependency_overrides.clear() assert response.status_code == 200 body = response.json() assert body["title"] == "Summer Deal" assert body["slug"] == "summer-deal" assert body["description"] is None assert body["coverImage"] == "https://cdn.example.test/summer.webp" assert body["priceAmount"] == 57500 assert body["priceUnit"] == "起/人" assert body["tags"] == ["无可比拟", "尊享礼遇"] assert body["status"] == "published" assert body["startsAt"] == "2026-07-01T08:00:00" assert body["endsAt"] is None assert "image" not in body assert "isActive" not in body assert "sortOrder" not in body assert campaign.status == "published" assert campaign.priceAmount == 57500 assert campaign.priceUnit == "起/人" assert campaign.tags == ["无可比拟", "尊享礼遇"] assert fake_db.added[-1].action == "update" assert fake_db.added[-1].entity == "campaign" def test_site_config_patch_route_section_updates_copy_and_product_order(): section = make_route_section() product_a = make_product(id="product-a") product_b = make_product(id="product-b") fake_db = FakeDb( get_result=section, scalar_results=[[product_b, product_a], []], execute_results=[[]], ) app = authenticated_app(fake_db) try: response = TestClient(app).patch( "/api/admin/site-config/routeSections/routes", json={ "title": " 新经典线路 ", "subtitle": " 新副文案 ", "isActive": False, "productIds": ["product-b", "product-a"], }, ) finally: app.dependency_overrides.clear() assert response.status_code == 200 body = response.json() assert body["title"] == "新经典线路" assert body["subtitle"] == "新副文案" assert body["isActive"] is False assert body["productIds"] == ["product-b", "product-a"] links = [item for item in fake_db.added if isinstance(item, RouteSectionProduct)] assert [(link.productId, link.sortOrder) for link in links] == [("product-b", 0), ("product-a", 1)] assert fake_db.added[-1].entity == "route_section" assert fake_db.committed def test_site_config_patch_route_section_rejects_product_used_by_another_section(): section = make_route_section(id="routes") conflict = make_route_section_product(sectionId="route-section-other", productId="product-a") fake_db = FakeDb( get_result=section, scalar_results=[[make_product(id="product-a")], [conflict]], ) app = authenticated_app(fake_db) try: response = TestClient(app).patch( "/api/admin/site-config/routeSections/routes", json={"productIds": ["product-a"]}, ) finally: app.dependency_overrides.clear() assert response.status_code == 409 assert response.json()["code"] == "ROUTE_SECTION_PRODUCT_CONFLICT" assert not fake_db.committed def test_public_site_config_campaigns_include_price_and_tags(): campaign = make_campaign(status="published", tags=["自然奇景", "小众秘境"]) fake_db = FakeDb(scalar_results=[[], [], [], [], [], [], [], [campaign], [], [], []]) result = site_config(fake_db, active_only=True) assert result["campaigns"] == [ { "id": "campaign-test", "slug": "classic-deal", "title": "经典打卡特惠", "description": "经典首游活动", "coverImage": "https://cdn.example.test/campaign.webp", "priceAmount": 162500, "priceUnit": "起/人", "tags": ["自然奇景", "小众秘境"], "status": "published", "startsAt": None, "endsAt": None, "createdAt": "2026-01-01T00:00:00", "updatedAt": "2026-01-02T00:00:00", } ] def test_public_site_config_route_sections_filter_unpublished_products(): published = make_product(id="product-published", status="published") draft = make_product(id="product-draft", status="draft") section = make_route_section( products=[ make_route_section_product(product=published, productId=published.id, sortOrder=0), make_route_section_product(product=draft, productId=draft.id, sortOrder=1), ] ) fake_db = FakeDb(scalar_results=[[], [], [], [], [], [], [], [], [section], [], []]) result = site_config(fake_db, active_only=True) assert result["routeSections"] == [ { "id": "routes", "title": "经典人文打卡线路", "subtitle": "黄果树、荔波小七孔、千户苗寨、镇远古城、梵净山一次串联", "productIds": ["product-published"], "isActive": True, } ] def test_public_site_config_includes_hotel_and_vehicle_modules(): hotel_group = make_hotel_group() vehicle_option = make_vehicle_option() fake_db = FakeDb(scalar_results=[[], [], [], [], [], [], [], [], [], [hotel_group], [vehicle_option]]) result = site_config(fake_db, active_only=True) assert result["hotelGroups"][0]["title"] == "经典酒店" assert result["hotelGroups"][0]["description"] == "经典城市酒店与山野度假住宿组合" assert result["hotelGroups"][0]["coverImage"] == "/assets/hotel.jpg" assert result["hotelGroups"][0]["priceAmount"] == 68000 assert result["hotelGroups"][0]["priceUnit"] == "起/晚" assert result["hotelGroups"][0]["tags"] == ["温泉", "亲子"] assert result["hotelGroups"][0]["status"] == "published" assert result["vehicleOptions"][0]["title"] == "5座舒适用车" assert result["vehicleOptions"][0]["image"] == "/assets/vehicle.jpg" 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 @pytest.mark.parametrize( "item_ids", [ ["route-section-a", "route-section-a", "route-section-b"], ["route-section-a", "route-section-b"], ["route-section-a", "route-section-b", "route-section-x"], ], ) def test_site_config_reorder_route_sections_rejects_duplicate_missing_and_unknown_ids(item_ids): items = [ make_route_section(id="route-section-a", title="A", sortOrder=0), make_route_section(id="route-section-b", title="B", sortOrder=1), make_route_section(id="route-section-c", title="C", sortOrder=2), ] fake_db = FakeDb(scalar_results=[items]) app = authenticated_app(fake_db) try: response = TestClient(app).patch("/api/admin/site-config/routeSections/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_site_config_reorder_map_is_not_supported(): fake_db = FakeDb() app = authenticated_app(fake_db) try: response = TestClient(app).patch("/api/admin/site-config/map/reorder", json={"itemIds": ["map-test"]}) finally: app.dependency_overrides.clear() assert response.status_code == 400 assert response.json()["code"] == "MODULE_CONFIG_REORDER_UNSUPPORTED" assert not fake_db.committed def test_site_config_reorder_campaigns_is_not_supported(): fake_db = FakeDb() app = authenticated_app(fake_db) try: response = TestClient(app).patch("/api/admin/site-config/campaigns/reorder", json={"itemIds": ["campaign-test"]}) finally: app.dependency_overrides.clear() assert response.status_code == 400 assert response.json()["code"] == "MODULE_CONFIG_REORDER_UNSUPPORTED" assert not fake_db.committed def test_site_config_reorder_route_sections_reassigns_sort_order(): items = [ make_route_section(id="route-section-a", title="经典", sortOrder=0), make_route_section(id="route-section-b", title="户外", sortOrder=1), make_route_section(id="route-section-c", title="混搭", sortOrder=2), ] fake_db = FakeDb(scalar_results=[items]) app = authenticated_app(fake_db) try: response = TestClient(app).patch( "/api/admin/site-config/routeSections/reorder", json={"itemIds": ["route-section-b", "route-section-a", "route-section-c"]}, ) finally: app.dependency_overrides.clear() assert response.status_code == 200 assert {item.id: item.sortOrder for item in items} == {"route-section-b": 0, "route-section-a": 1, "route-section-c": 2} assert [item["id"] for item in response.json()["items"]] == ["route-section-b", "route-section-a", "route-section-c"] assert fake_db.added[-1].entity == "route_section" assert fake_db.committed def test_site_config_delete_route_section_returns_json_reorders_and_does_not_delete_products(): section = make_route_section(id="route-section-delete", title="待删除", sortOrder=1) remaining_a = make_route_section(id="route-section-a", title="保留 A", sortOrder=0) remaining_b = make_route_section(id="route-section-b", title="保留 B", sortOrder=2) fake_db = FakeDb(get_result=section, scalar_results=[[remaining_a, remaining_b]]) app = authenticated_app(fake_db) try: response = TestClient(app).delete("/api/admin/site-config/routeSections/route-section-delete") finally: app.dependency_overrides.clear() assert response.status_code == 200 assert response.json() == {"id": "route-section-delete"} assert fake_db.deleted == [section] assert {item.id: item.sortOrder for item in [remaining_a, remaining_b]} == {"route-section-a": 0, "route-section-b": 1} assert fake_db.added[-1].action == "delete" assert fake_db.added[-1].entity == "route_section" assert fake_db.committed def test_site_config_delete_map_image_returns_json_and_audits_without_reorder(): map_image = make_map_image() fake_db = FakeDb(get_result=map_image) app = authenticated_app(fake_db) try: response = TestClient(app).delete("/api/admin/site-config/map/map-test") finally: app.dependency_overrides.clear() assert response.status_code == 200 assert response.json() == {"id": "map-test"} assert fake_db.deleted == [map_image] assert fake_db.added[-1].action == "delete" assert fake_db.added[-1].entity == "map_image" assert fake_db.committed def test_site_config_delete_campaign_returns_json_and_audits_without_reorder(): campaign = make_campaign() fake_db = FakeDb(get_result=campaign) app = authenticated_app(fake_db) try: response = TestClient(app).delete("/api/admin/site-config/campaigns/campaign-test") finally: app.dependency_overrides.clear() assert response.status_code == 200 assert response.json() == {"id": "campaign-test"} assert fake_db.deleted == [campaign] assert fake_db.added[-1].action == "delete" assert fake_db.added[-1].entity == "campaign" assert 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"