WonderQ-Admin/tests/test_api_contracts.py

from datetime import datetime

import pytest
from fastapi.testclient import TestClient
from pydantic import ValidationError

from app import serializers
from app.auth import hash_password, require_admin
from app.database import get_db
from app.main import create_app
from app.models import AdminUser, CtaBanner, Destination, DestinationAlias, HeroSlide, Lead, MediaAsset, Product, ProductImage, ThemeCard
from app.routers import admin as admin_router
from app.routers.admin import normalize_detail_sections, normalize_images
from app.schemas import (
    AdminProductQuery,
    LeadCreateIn,
    LeadQuery,
    ProductCreateIn,
    ProductImageIn,
    ProductQuery,
)


class FakeScalarResult:
    def __init__(self, items):
        self.items = items

    def unique(self):
        return self

    def all(self):
        return self.items

    def first(self):
        return self.items[0] if self.items else None

    def one(self):
        return self.items[0]


class FakeExecuteResult:
    def __init__(self, rows):
        self.rows = rows

    def all(self):
        return self.rows


class FakeDb:
    def __init__(self, *, scalar_results=None, scalar_values=None, execute_results=None, get_result=None):
        self.scalar_results = list(scalar_results or [])
        self.scalar_values = list(scalar_values or [])
        self.execute_results = list(execute_results or [])
        self.get_result = get_result
        self.added = []
        self.deleted = []
        self.committed = False

    def scalars(self, _stmt):
        return FakeScalarResult(self.scalar_results.pop(0))

    def scalar(self, _stmt):
        return self.scalar_values.pop(0)

    def execute(self, _stmt):
        return FakeExecuteResult(self.execute_results.pop(0))

    def get(self, _model, _item_id):
        return self.get_result

    def add(self, item):
        self.added.append(item)

    def delete(self, item):
        self.deleted.append(item)

    def flush(self):
        for item in self.added:
            if not getattr(item, "id", None):
                item.id = f"{item.__class__.__name__.lower()}-test-id"

    def commit(self):
        self.committed = True

    def refresh(self, item):
        if not getattr(item, "id", None):
            item.id = "lead-test-id"


def make_destination(**overrides):
    destination = Destination(
        id=overrides.get("id", "dest-test"),
        name=overrides.get("name", "测试目的地"),
        slug=overrides.get("slug", "test-destination"),
        region=overrides.get("region", "测试区域"),
        image=overrides.get("image", "/assets/test.jpg"),
        isHot=overrides.get("isHot", True),
        sortOrder=overrides.get("sortOrder", 1),
        isActive=overrides.get("isActive", True),
        createdAt=datetime(2026, 1, 1),
        updatedAt=datetime(2026, 1, 2),
    )
    destination.aliases = [
        DestinationAlias(id="alias-b", alias="别名B", destinationId=destination.id),
        DestinationAlias(id="alias-a", alias="别名A", destinationId=destination.id),
    ]
    destination.products = []
    return destination


def make_product(**overrides):
    destination = overrides.get("destination", make_destination())
    product = Product(
        id=overrides.get("id", "product-test"),
        sourceId=overrides.get("sourceId", 101),
        title=overrides.get("title", "测试线路"),
        subtitle=overrides.get("subtitle", "测试副标题"),
        destinationId=destination.id,
        priceAmount=overrides.get("priceAmount", 1000),
        priceUnit=overrides.get("priceUnit", "起/人"),
        tags=overrides.get("tags", ["测试", "线路"]),
        coverImage=overrides.get("coverImage", "/assets/cover.jpg"),
        summary=overrides.get("summary", "测试摘要"),
        detailSections=overrides.get("detailSections", None),
        status=overrides.get("status", "published"),
        sortWeight=overrides.get("sortWeight", 1),
        publishedAt=datetime(2026, 1, 3),
        createdAt=datetime(2026, 1, 1),
        updatedAt=datetime(2026, 1, 2),
    )
    product.destination = destination
    product.images = [
        ProductImage(id="img-b", productId=product.id, url="/assets/b.jpg", alt="B", sortOrder=2),
        ProductImage(id="img-a", productId=product.id, url="/assets/a.jpg", alt="A", sortOrder=1),
    ]
    return product


def make_hero_slide(**overrides):
    return HeroSlide(
        id=overrides.get("id", "slide-test"),
        title=overrides.get("title", "测试轮播"),
        kicker=overrides.get("kicker", "测试副标题"),
        image=overrides.get("image", "/assets/slide.jpg"),
        targetType=overrides.get("targetType", "campaign"),
        targetValue=overrides.get("targetValue", "campaign-test"),
        sortOrder=overrides.get("sortOrder", 1),
        isActive=overrides.get("isActive", True),
        createdAt=datetime(2026, 1, 1),
        updatedAt=datetime(2026, 1, 2),
    )


def make_theme_card(**overrides):
    return ThemeCard(
        id=overrides.get("id", "theme-test"),
        label=overrides.get("label", "测试主题"),
        image=overrides.get("image", "/assets/theme.jpg"),
        targetType=overrides.get("targetType", "search"),
        targetValue=overrides.get("targetValue", "测试主题"),
        sortOrder=overrides.get("sortOrder", 1),
        isActive=overrides.get("isActive", True),
        createdAt=datetime(2026, 1, 1),
        updatedAt=datetime(2026, 1, 2),
    )


def make_cta_banner(**overrides):
    return CtaBanner(
        id=overrides.get("id", "cta-test"),
        alt=overrides.get("alt", "测试运营入口"),
        image=overrides.get("image", "/assets/cta.jpg"),
        targetType=overrides.get("targetType", "lead"),
        targetValue=overrides.get("targetValue", "cta-test"),
        sortOrder=overrides.get("sortOrder", 1),
        isActive=overrides.get("isActive", True),
        createdAt=datetime(2026, 1, 1),
        updatedAt=datetime(2026, 1, 2),
    )


def make_admin_user():
    return AdminUser(
        id="admin-test",
        email="admin@example.test",
        name="Admin",
        role="admin",
        passwordHash="not-used",
        isActive=True,
    )


def authenticated_app(fake_db):
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db
    app.dependency_overrides[require_admin] = make_admin_user
    return app


def test_public_product_serializer_limits_destination_and_stabilizes_arrays():
    data = serializers.public_product_dict(make_product())

    assert data["destination"] == {"id": "dest-test", "name": "测试目的地"}
    assert [image["id"] for image in data["images"]] == ["img-a", "img-b"]
    assert data["detailSections"] == []


def test_destination_serializer_uses_precomputed_product_count_and_sorts_aliases():
    destination = make_destination()

    data = serializers.destination_dict(destination, include_count=True, product_count=7)

    assert data["_count"] == {"products": 7}
    assert [alias["id"] for alias in data["aliases"]] == ["alias-a", "alias-b"]


def test_lead_schema_accepts_date_only_string():
    lead = LeadCreateIn(phone="  contact   handle  ", travelDate="2027-01-01")

    assert lead.phone == "contact handle"
    assert lead.travelDate == datetime(2027, 1, 1)


def test_query_status_validation_rejects_unknown_values():
    with pytest.raises(ValidationError):
        ProductQuery(status="hidden")
    with pytest.raises(ValidationError):
        AdminProductQuery(status="hidden")
    with pytest.raises(ValidationError):
        LeadQuery(status="closed")


def test_normalize_images_reassigns_sort_order_by_payload_order():
    images = [
        ProductImageIn(url="/assets/first.jpg", sortOrder=99),
        ProductImageIn(url="/assets/second.jpg", sortOrder=10),
    ]

    assert [image["sortOrder"] for image in normalize_images(images)] == [0, 1]


def test_empty_detail_sections_are_accepted_and_dropped():
    body = ProductCreateIn(
        title="测试线路",
        detailSections=[{"key": "", "label": "", "blocks": []}],
    )

    assert normalize_detail_sections(body.detailSections) == []


def test_public_products_endpoint_uses_public_product_contract():
    product = make_product()
    fake_db = FakeDb(scalar_results=[[product]])
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db

    try:
        response = TestClient(app).get("/api/public/products")
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    item = response.json()["items"][0]
    assert item["destination"] == {"id": "dest-test", "name": "测试目的地"}
    assert item["images"][0]["url"] == "/assets/a.jpg"


def test_public_leads_endpoint_accepts_date_only_and_returns_minimal_response():
    fake_db = FakeDb()
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db

    try:
        response = TestClient(app).post(
            "/api/public/leads",
            json={
                "phone": "  contact   handle  ",
                "travelDate": "2027-01-01",
                "peopleCount": 2,
                "budgetMin": 0,
                "budgetMax": 100,
            },
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 201
    assert response.json() == {"id": "lead-test-id", "status": "new"}
    assert fake_db.added[0].phone == "contact handle"
    assert fake_db.added[0].travelDate == datetime(2027, 1, 1)


def test_admin_destinations_endpoint_uses_precomputed_product_counts():
    destination = make_destination()
    fake_db = FakeDb(scalar_results=[[destination]], execute_results=[[(destination.id, 7)]])
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db
    app.dependency_overrides[require_admin] = lambda: AdminUser(
        id="admin-test",
        email="admin@example.test",
        name="Admin",
        role="admin",
        passwordHash="not-used",
        isActive=True,
    )

    try:
        response = TestClient(app).get("/api/admin/destinations")
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    assert response.json()["items"][0]["_count"] == {"products": 7}


def test_admin_requires_auth_for_protected_endpoint():
    response = TestClient(create_app()).get("/api/admin/products")

    assert response.status_code == 401
    assert response.json() == {"message": "请先登录后台"}


def test_admin_login_returns_token_and_user():
    admin_user = AdminUser(
        id="admin-test",
        email="admin@example.com",
        name="Admin",
        role="admin",
        passwordHash=hash_password("ChangeMe123!", rounds=4),
        isActive=True,
    )
    fake_db = FakeDb(scalar_values=[admin_user])
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db

    try:
        response = TestClient(app).post(
            "/api/admin/auth/login",
            json={"email": "admin@example.com", "password": "ChangeMe123!"},
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    body = response.json()
    assert body["token"]
    assert body["user"] == {
        "id": "admin-test",
        "email": "admin@example.com",
        "name": "Admin",
        "role": "admin",
    }


def test_site_config_patch_ignores_fields_not_allowed_for_module_and_audits():
    destination = make_destination()
    fake_db = FakeDb(get_result=destination)
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db
    app.dependency_overrides[require_admin] = lambda: AdminUser(
        id="admin-test",
        email="admin@example.test",
        name="Admin",
        role="admin",
        passwordHash="not-used",
        isActive=True,
    )

    try:
        response = TestClient(app).patch(
            "/api/admin/site-config/destinations/dest-test",
            json={"name": "新目的地", "targetType": "ignored"},
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    assert response.json()["name"] == "新目的地"
    assert not hasattr(destination, "targetType")
    assert fake_db.committed
    assert fake_db.added[-1].entity == "destination"


@pytest.mark.parametrize(
    ("module", "payload", "expected"),
    [
        ("heroSlides", {"title": "新轮播", "image": None, "targetType": None}, {"title": "新轮播", "image": None}),
        ("destinations", {"name": "新目的地", "slug": "", "region": None, "isHot": True}, {"name": "新目的地", "slug": "e696b0e79baee79a84e59cb0", "isHot": True}),
        ("themes", {"label": "新主题", "image": None}, {"label": "新主题", "image": ""}),
        ("ctaBanners", {"alt": "新运营入口", "image": None, "targetType": None}, {"alt": "新运营入口", "image": "", "targetType": ""}),
    ],
)
def test_site_config_create_modules_defaults_fields_and_audits(module, payload, expected):
    fake_db = FakeDb(scalar_values=[4])
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).post(f"/api/admin/site-config/{module}", json=payload)
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 201
    body = response.json()
    for key, value in expected.items():
        assert body[key] == value
    assert body["isActive"] is True
    assert body["sortOrder"] == 5
    assert fake_db.committed
    assert fake_db.added[-1].action == "create"


def test_site_config_invalid_module_returns_structured_error():
    fake_db = FakeDb()
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).post("/api/admin/site-config/unknown", json={"title": "测试"})
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 400
    assert response.json() == {
        "message": "模块不存在或无权限操作",
        "code": "MODULE_CONFIG_FORBIDDEN",
        "details": {"module": "unknown"},
    }


def test_site_config_create_requires_module_primary_field():
    fake_db = FakeDb()
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).post("/api/admin/site-config/themes", json={"image": "/assets/theme.jpg"})
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 422
    assert response.json()["code"] == "MODULE_CONFIG_VALIDATION_ERROR"
    assert response.json()["details"] == {"field": "label"}


def test_admin_site_config_hero_slides_use_dedicated_contract_without_targets():
    hero_slide = make_hero_slide(targetType="campaign", targetValue="campaign-test")
    fake_db = FakeDb(scalar_results=[[], [hero_slide], [], []])
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).get("/api/admin/site-config")
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    hero = response.json()["heroSlides"][0]
    assert hero["title"] == "测试轮播"
    assert hero["image"] == "/assets/slide.jpg"
    assert "targetType" not in hero
    assert "targetValue" not in hero


def test_site_config_create_hero_slide_ignores_target_fields_and_returns_dedicated_contract():
    fake_db = FakeDb(scalar_values=[0])
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).post(
            "/api/admin/site-config/heroSlides",
            json={
                "title": "  新轮播  ",
                "kicker": "",
                "image": None,
                "targetType": "campaign",
                "targetValue": "ignored",
            },
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 201
    body = response.json()
    assert body["title"] == "新轮播"
    assert body["image"] is None
    assert "targetType" not in body
    assert "targetValue" not in body
    created = fake_db.added[0]
    assert created.targetType is None
    assert created.targetValue is None


def test_site_config_patch_hero_slide_ignores_target_fields_and_returns_dedicated_contract():
    hero_slide = make_hero_slide(targetType="campaign", targetValue="campaign-test")
    fake_db = FakeDb(get_result=hero_slide)
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).patch(
            "/api/admin/site-config/heroSlides/slide-test",
            json={
                "title": "夏日贵州小包团",
                "image": None,
                "targetType": "search",
                "targetValue": "ignored",
            },
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    body = response.json()
    assert body["title"] == "夏日贵州小包团"
    assert body["image"] is None
    assert "targetType" not in body
    assert "targetValue" not in body
    assert hero_slide.targetType == "campaign"
    assert hero_slide.targetValue == "campaign-test"


def test_site_config_patch_updates_destination_contract_fields():
    destination = make_destination()
    fake_db = FakeDb(get_result=destination)
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).patch(
            "/api/admin/site-config/destinations/dest-test",
            json={"slug": "new-slug", "region": None, "image": None, "isHot": False, "sortOrder": 8, "targetType": "ignored"},
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    body = response.json()
    assert body["slug"] == "new-slug"
    assert body["region"] is None
    assert body["image"] is None
    assert body["isHot"] is False
    assert body["sortOrder"] == 8
    assert not hasattr(destination, "targetType")
    assert fake_db.added[-1].action == "update"


def test_site_config_delete_returns_json_reorders_remaining_items_and_audits():
    delete_item = make_theme_card(id="theme-delete", sortOrder=1)
    remaining = [make_theme_card(id="theme-b", sortOrder=5), make_theme_card(id="theme-a", sortOrder=9)]
    fake_db = FakeDb(get_result=delete_item, scalar_results=[remaining])
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).delete("/api/admin/site-config/themes/theme-delete")
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    assert response.json() == {"id": "theme-delete"}
    assert fake_db.deleted == [delete_item]
    assert [item.sortOrder for item in remaining] == [0, 1]
    assert fake_db.committed
    assert fake_db.added[-1].action == "delete"
    assert fake_db.added[-1].entity == "theme_card"


def test_site_config_delete_destination_with_products_returns_conflict():
    destination = make_destination()
    fake_db = FakeDb(get_result=destination, scalar_values=[1])
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).delete("/api/admin/site-config/destinations/dest-test")
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 409
    assert response.json()["code"] == "MODULE_CONFIG_CONFLICT"
    assert not fake_db.deleted


def test_site_config_reorder_reassigns_sort_order_and_returns_items():
    items = [
        make_hero_slide(id="slide-1", title="第一张", sortOrder=0),
        make_hero_slide(id="slide-2", title="第二张", sortOrder=1),
        make_hero_slide(id="slide-3", title="第三张", sortOrder=2),
    ]
    fake_db = FakeDb(scalar_results=[items])
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).patch(
            "/api/admin/site-config/heroSlides/reorder",
            json={"itemIds": ["slide-2", "slide-1", "slide-3"]},
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    assert [item.id for item in items] == ["slide-1", "slide-2", "slide-3"]
    assert {item.id: item.sortOrder for item in items} == {"slide-2": 0, "slide-1": 1, "slide-3": 2}
    assert [item["id"] for item in response.json()["items"]] == ["slide-2", "slide-1", "slide-3"]
    assert [item["sortOrder"] for item in response.json()["items"]] == [0, 1, 2]
    assert fake_db.committed
    assert fake_db.added[-1].action == "reorder"


@pytest.mark.parametrize(
    "item_ids",
    [
        ["slide-1", "slide-1", "slide-2"],
        ["slide-1", "slide-2"],
        ["slide-1", "slide-2", "other-module-id"],
    ],
)
def test_site_config_reorder_rejects_duplicate_missing_and_unknown_ids(item_ids):
    items = [
        make_hero_slide(id="slide-1"),
        make_hero_slide(id="slide-2"),
        make_hero_slide(id="slide-3"),
    ]
    fake_db = FakeDb(scalar_results=[items])
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).patch("/api/admin/site-config/heroSlides/reorder", json={"itemIds": item_ids})
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 400
    assert response.json()["code"] == "MODULE_CONFIG_REORDER_INVALID"
    assert not fake_db.committed


def test_admin_media_upload_streams_image_to_oss_records_asset_and_audits(monkeypatch):
    uploaded = {}

    def fake_upload(file_obj, key, mime_type, size_bytes):
        uploaded["key"] = key
        uploaded["mimeType"] = mime_type
        uploaded["sizeBytes"] = size_bytes
        uploaded["body"] = file_obj.read()
        return f"https://cdn.example.test/{key}"

    monkeypatch.setattr(admin_router, "upload_image_to_oss", fake_upload)
    fake_db = FakeDb()
    app = authenticated_app(fake_db)
    content = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR"

    try:
        response = TestClient(app).post(
            "/api/admin/media-assets/upload",
            data={"group": "heroSlides"},
            files={"file": ("hero.png", content, "image/png")},
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 201
    body = response.json()
    assert body["url"].startswith("https://cdn.example.test/admin/heroSlides/")
    assert body["name"] == "hero.png"
    assert body["mimeType"] == "image/png"
    assert body["sizeBytes"] == len(content)
    assert body["group"] == "heroSlides"
    assert uploaded["body"] == content
    assert uploaded["key"].endswith(".png")
    assert isinstance(fake_db.added[0], MediaAsset)
    assert fake_db.added[-1].entity == "media_asset"
    assert fake_db.committed


def test_admin_media_upload_rejects_non_image_file():
    fake_db = FakeDb()
    app = authenticated_app(fake_db)

    try:
        response = TestClient(app).post(
            "/api/admin/media-assets/upload",
            data={"group": "docs"},
            files={"file": ("note.txt", b"hello", "text/plain")},
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 400
    assert response.json()["code"] == "MEDIA_UPLOAD_INVALID_TYPE"
    assert not fake_db.added
    assert not fake_db.committed


def test_admin_lead_status_update_returns_updated_status_and_audits():
    lead = Lead(
        id="lead-test",
        destination="测试目的地",
        phone="contact handle",
        status="new",
        createdAt=datetime(2026, 1, 1),
        updatedAt=datetime(2026, 1, 2),
    )
    fake_db = FakeDb(get_result=lead)
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db
    app.dependency_overrides[require_admin] = lambda: AdminUser(
        id="admin-test",
        email="admin@example.test",
        name="Admin",
        role="admin",
        passwordHash="not-used",
        isActive=True,
    )

    try:
        response = TestClient(app).patch("/api/admin/leads/lead-test/status", json={"status": "contacted"})
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    assert response.json()["status"] == "contacted"
    assert lead.status == "contacted"
    assert fake_db.committed
    assert fake_db.added[-1].entity == "lead"