WonderQ-Project/WonderQ-Admin/tests/test_api_contracts.py

from datetime import datetime

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

from app.auth import create_customer_token, hash_password, require_admin
from app.database import get_db
from app.main import create_app
from app.models import AdminUser, Customer, HeroSlide, Lead, MediaAsset, VehicleOption
from app.routers import admin as admin_router
from app.routers import public as public_router
from app.routers.shared import site_config
from app.redis_session import InMemoryAdminSessionStore, get_admin_session_store
from app.schemas import LeadCreateIn, LeadQuery


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_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"),
        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_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_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_retired_domain_endpoints_are_not_registered():
    client = TestClient(create_app())

    assert client.get("/api/public/products").status_code == 404
    assert client.get("/api/public/products/retired").status_code == 404
    assert client.get("/api/admin/products").status_code == 404
    assert client.get("/api/public/destinations").status_code == 404
    assert client.get("/api/admin/destinations").status_code == 404


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):
        LeadQuery(status="closed")


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() == {
        "code": 201,
        "msg": "success",
        "data": {"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_public_phone_login_creates_customer_and_returns_masked_session(monkeypatch):
    monkeypatch.setattr(public_router, "exchange_phone_code", lambda code: "10000000000", raising=False)
    fake_db = FakeDb(scalar_values=[None])
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db

    try:
        response = TestClient(app).post("/api/public/auth/phone-login", json={"code": "phone-code"})
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    body = response.json()["data"]
    assert body["token"]
    assert body["customer"] == {"id": "customer-test-id", "phoneMasked": "100****0000"}
    assert fake_db.added[0].phone == "10000000000"
    assert fake_db.committed


def test_public_phone_login_requires_wechat_configuration(monkeypatch):
    def missing_config(_code):
        raise public_router.WechatConfigError("微信小程序登录未配置")

    monkeypatch.setattr(public_router, "exchange_phone_code", missing_config, raising=False)
    app = create_app()
    app.dependency_overrides[get_db] = lambda: FakeDb()

    try:
        response = TestClient(app).post("/api/public/auth/phone-login", json={"code": "phone-code"})
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 503
    assert response.json() == {
        "code": 503,
        "msg": "微信小程序登录未配置",
        "data": None,
    }


def test_public_me_returns_customer_for_customer_token():
    customer = Customer(id="customer-test", phone="10000000000")
    fake_db = FakeDb(get_result=customer)
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db
    token = create_customer_token(customer)

    try:
        response = TestClient(app).get("/api/public/auth/me", headers={"Authorization": f"Bearer {token}"})
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 200
    assert response.json() == {
        "code": 200,
        "msg": "success",
        "data": {"id": "customer-test", "phoneMasked": "100****0000"},
    }


def test_public_me_requires_customer_token():
    response = TestClient(create_app()).get("/api/public/auth/me")

    assert response.status_code == 401
    assert response.json() == {"code": 401, "msg": "请先登录", "data": None}


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

    assert response.status_code == 401
    assert response.json() == {"code": 401, "msg": "请先登录后台", "data": None}


def test_admin_dynamic_routers_requires_admin_authentication():
    response = TestClient(create_app()).get("/api/admin/system/routers")

    assert response.status_code == 401
    assert response.json() == {"code": 401, "msg": "请先登录后台", "data": None}


def test_admin_login_requires_valid_captcha():
    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])
    store = InMemoryAdminSessionStore()
    store.create_captcha("captcha-test", "ABCD", 120)
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db
    app.dependency_overrides[get_admin_session_store] = lambda: store

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

    assert response.status_code == 401
    assert response.json()["data"] is None


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])
    store = InMemoryAdminSessionStore()
    store.create_captcha("captcha-contract", "ABCD", 120)
    app = create_app()
    app.dependency_overrides[get_db] = lambda: fake_db
    app.dependency_overrides[get_admin_session_store] = lambda: store

    try:
        response = TestClient(app).post(
            "/api/admin/auth/login",
            json={
                "email": "admin@example.com",
                "password": "ChangeMe123!",
                "captchaId": "captcha-contract",
                "captchaCode": "ABCD",
                "rememberMe": True,
            },
        )
    finally:
        app.dependency_overrides.clear()

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


@pytest.mark.parametrize(
    ("module", "payload", "expected"),
    [
        ("heroSlides", {"title": "新轮播", "image": None, "targetType": None}, {"title": "新轮播", "image": None}),
        ("vehicleOptions", {"title": "5座舒适用车", "description": "用车文案", "image": None}, {"title": "5座舒适用车", "description": "用车文案", "image": None}),
    ],
)
def test_site_config_create_supported_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()["data"]
    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() == {
        "code": 400,
        "msg": "模块不存在或无权限操作",
        "data": None,
        "errorCode": "MODULE_CONFIG_FORBIDDEN",
        "details": {"module": "unknown"},
    }


@pytest.mark.parametrize(
    ("module", "payload"),
    [
        ("themes", {"label": "已移除模块"}),
        ("map", {"image": "/assets/removed-map.webp"}),
        ("destinationRegions", {"title": "已移除热门区域"}),
    ],
)
def test_site_config_create_rejects_removed_module(module, payload):
    fake_db = FakeDb()
    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 == 400
    assert response.json()["errorCode"] == "MODULE_CONFIG_FORBIDDEN"


def test_admin_site_config_hero_slides_use_dedicated_contract_without_targets():
    hero_slide = make_hero_slide(targetType="search", targetValue="search-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()["data"]["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": "search",
                "targetValue": "ignored",
            },
        )
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 201
    body = response.json()["data"]
    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="search", targetValue="search-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()["data"]
    assert body["title"] == "夏日贵州小包团"
    assert body["image"] is None
    assert "targetType" not in body
    assert "targetValue" not in body
    assert hero_slide.targetType == "search"
    assert hero_slide.targetValue == "search-test"


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()["data"]["items"]] == ["slide-2", "slide-1", "slide-3"]
    assert [item["sortOrder"] for item in response.json()["data"]["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()["errorCode"] == "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()["data"]
    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()["errorCode"] == "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() == {
        "code": 200,
        "msg": "success",
        "data": {"items": [], "total": 0, "offset": 0, "take": 50},
    }
    stmt = str(fake_db.scalar_statements[0])
    assert '"Lead".status' in stmt
    assert '"Lead"."sourcePage"' in stmt
    assert '"Lead"."createdAt"' 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()["data"]["status"] == "contacted"
    assert lead.status == "contacted"
    assert fake_db.committed
    assert fake_db.added[-1].entity == "lead"