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, DemandFeatureCard, DemandForm, DemandHero, DemandRecommendation, DemandRecommendationProduct, 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.scalar_statements = [] self.execute_statements = [] self.get_result = get_result self.added = [] self.deleted = [] self.committed = False def scalars(self, stmt): self.scalar_statements.append(stmt) return FakeScalarResult(self.scalar_results.pop(0) if self.scalar_results else []) def scalar(self, stmt): self.scalar_statements.append(stmt) return self.scalar_values.pop(0) if self.scalar_values else None def execute(self, stmt): self.execute_statements.append(stmt) return FakeExecuteResult(self.execute_results.pop(0) if self.execute_results else []) 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_demand_hero(**overrides): return DemandHero( id=overrides.get("id", "demand-hero-test"), title=overrides.get("title", "告诉我们日期、人数和想法"), kicker=overrides.get("kicker", "3步定制"), description=overrides.get("description", "管家会按同行人、预算和体力强度，重新组合酒店、用车和景点节奏。"), steps=overrides.get("steps", ["提交需求", "管家沟通", "确认方案"]), sortOrder=overrides.get("sortOrder", 0), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_demand_feature_card(**overrides): return DemandFeatureCard( id=overrides.get("id", "demand-feature-card-test"), title=overrides.get("title", "动线"), description=overrides.get("description", "按天数顺路排"), sortOrder=overrides.get("sortOrder", 0), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_demand_form(**overrides): return DemandForm( id=overrides.get("id", "demand-form-test"), destinationLabel=overrides.get("destinationLabel", "目的地/玩法"), destinationPlaceholder=overrides.get("destinationPlaceholder", "例如：贵州、黄果树、西江苗寨"), phoneLabel=overrides.get("phoneLabel", "联系方式"), phonePlaceholder=overrides.get("phonePlaceholder", "手机号 / 微信号"), noteLabel=overrides.get("noteLabel", "补充说明"), notePlaceholder=overrides.get("notePlaceholder", "出行日期、人数、酒店偏好、预算范围"), submitLabel=overrides.get("submitLabel", "提交出行需求"), chips=overrides.get("chips", ["贵州", "黄果树", "荔波小七孔", "西江苗寨", "梵净山", "万峰林"]), isActive=overrides.get("isActive", True), createdAt=datetime(2026, 1, 1), updatedAt=datetime(2026, 1, 2), ) def make_demand_recommendation(**overrides): item = DemandRecommendation( id=overrides.get("id", "demand-recommendation-test"), 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), ) item.products = overrides.get("products", []) return item def make_demand_recommendation_product(**overrides): product = overrides.get("product", make_product(id=overrides.get("productId", "product-test"))) link = DemandRecommendationProduct( recommendationId=overrides.get("recommendationId", "demand-recommendation-test"), 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}), ( "demandHero", {"title": "需求页主视觉", "kicker": "3步定制", "description": "说明", "steps": ["提交", "沟通", "确认"]}, {"title": "需求页主视觉", "kicker": "3步定制", "description": "说明", "steps": ["提交", "沟通", "确认"]}, ), ( "demandFeatureCards", {"title": "动线", "description": "按天数顺路排"}, {"title": "动线", "description": "按天数顺路排"}, ), ( "demandForm", {"submitLabel": "提交出行需求", "destinationLabel": "目的地/玩法", "chips": [" 贵州 ", "", "黄果树"]}, {"submitLabel": "提交出行需求", "destinationLabel": "目的地/玩法", "chips": ["贵州", "黄果树"]}, ), ( "demandRecommendations", {"title": "热门推荐", "subtitle": "也可以先挑一条线路沟通"}, {"title": "热门推荐", "subtitle": "也可以先挑一条线路沟通", "productIds": []}, ), ], ) 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 if module == "demandForm": assert "sortOrder" not in body else: 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_admin_site_config_includes_demand_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["demandHero"] == [] assert body["demandFeatureCards"] == [] assert body["demandForm"] == [] assert body["demandRecommendations"] == [] 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_public_site_config_includes_demand_modules_and_filters_unpublished_recommendation_products(): published = make_product(id="product-published", status="published") draft = make_product(id="product-draft", status="draft") recommendation = make_demand_recommendation( products=[ make_demand_recommendation_product(product=published, productId=published.id, sortOrder=1), make_demand_recommendation_product(product=draft, productId=draft.id, sortOrder=0), ] ) fake_db = FakeDb( scalar_results=[ [], [], [], [], [], [], [], [], [], [], [], [make_demand_hero()], [make_demand_feature_card()], [make_demand_form()], [recommendation], ] ) result = site_config(fake_db, active_only=True) assert result["demandHero"][0]["title"] == "告诉我们日期、人数和想法" assert result["demandHero"][0]["steps"] == ["提交需求", "管家沟通", "确认方案"] assert result["demandFeatureCards"][0]["title"] == "动线" assert result["demandForm"][0]["submitLabel"] == "提交出行需求" assert result["demandForm"][0]["chips"] == ["贵州", "黄果树", "荔波小七孔", "西江苗寨", "梵净山", "万峰林"] assert result["demandRecommendations"] == [ { "id": "demand-recommendation-test", "title": "热门推荐", "subtitle": "也可以先挑一条线路沟通", "productIds": ["product-published"], "isActive": True, } ] 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_demand_form_rejects_duplicate_singleton(): fake_db = FakeDb(scalar_results=[[make_demand_form()]]) app = authenticated_app(fake_db) try: response = TestClient(app).post("/api/admin/site-config/demandForm", json={"submitLabel": "提交"}) finally: app.dependency_overrides.clear() assert response.status_code == 409 assert response.json()["code"] == "MODULE_CONFIG_SINGLETON_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_demand_recommendation_updates_product_order_without_route_conflict(): recommendation = make_demand_recommendation() product_a = make_product(id="product-a") product_b = make_product(id="product-b") fake_db = FakeDb( get_result=recommendation, scalar_results=[[product_b, product_a]], execute_results=[[]], ) app = authenticated_app(fake_db) try: response = TestClient(app).patch( "/api/admin/site-config/demandRecommendations/demand-recommendation-test", 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, DemandRecommendationProduct)] assert [(link.productId, link.sortOrder) for link in links] == [("product-b", 0), ("product-a", 1)] assert fake_db.added[-1].entity == "demand_recommendation" 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_leads_endpoint_accepts_source_keyword_and_created_range_filters(): fake_db = FakeDb(scalar_results=[[]]) app = authenticated_app(fake_db) try: response = TestClient(app).get( "/api/admin/leads", params={ "status": "new", "sourcePage": "demand_page", "keyword": "荔波", "createdFrom": "2026-01-01", "createdTo": "2026-01-31", "take": 50, }, ) finally: app.dependency_overrides.clear() assert response.status_code == 200 assert response.json() == {"items": []} stmt = str(fake_db.scalar_statements[0]) assert '"Lead".status' in stmt assert '"Lead"."sourcePage"' in stmt assert '"Lead"."createdAt"' in stmt assert '"Product".title' in stmt assert '"Lead".phone' in stmt assert '"Lead".destination' in stmt assert '"Lead".note' in stmt 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"