harness_zbt/backend/tests/integration/test_real_mysql_api.py

from collections.abc import Iterator
from datetime import UTC, datetime

import pytest
from fastapi.testclient import TestClient

from zbt.core.config import Settings
from zbt.main import create_app

settings = Settings()
TEST_DATABASES = {
    "insurance_s1_test_core",
    "insurance_s1_test_agent",
    "insurance_s1_test_analytics",
}
configured_databases = {
    settings.mysql_core_database,
    settings.mysql_agent_database,
    settings.mysql_analytics_database,
}

pytestmark = pytest.mark.skipif(
    settings.app_env != "test" or configured_databases != TEST_DATABASES,
    reason="真实MySQL API集成测试只允许连接三个白名单测试库",
)


@pytest.fixture(scope="module")
def client() -> Iterator[TestClient]:
    application = create_app(
        settings=settings,
        clock=lambda: datetime(2026, 7, 25, tzinfo=UTC),
    )
    with TestClient(application) as test_client:
        yield test_client
    engine = application.state.core_engine
    if engine is not None:
        engine.dispose()


def h5_login(client: TestClient) -> dict[str, str]:
    response = client.post(
        "/api/v1/h5/auth/login",
        json={"mobile": "18899990001", "code": "147258"},
    )
    assert response.status_code == 200
    return response.json()["data"]["tokens"]


def admin_login(client: TestClient, username: str) -> dict[str, str]:
    response = client.post(
        "/api/v1/admin/auth/login",
        json={"username": username, "password": "zaq1XSW@"},
    )
    assert response.status_code == 200
    return response.json()["data"]["tokens"]


def test_h5_authentication_and_current_catalog_use_real_mysql(
    client: TestClient,
) -> None:
    request_code = client.post(
        "/api/v1/h5/auth/request-code",
        json={"mobile": "18899990001"},
    )
    assert request_code.status_code == 200
    assert request_code.json()["data"] == {"expires_in": 300}
    assert "code" not in request_code.json()["data"]

    tokens = h5_login(client)
    headers = {"Authorization": f"Bearer {tokens['access_token']}"}
    profile = client.get("/api/v1/h5/me", headers=headers)
    assert profile.status_code == 200
    assert profile.json()["data"]["mobile_masked"] == "188****0001"

    products = client.get("/api/v1/h5/products", headers=headers)
    assert products.status_code == 200
    assert products.json()["data"]["total"] == 4
    assert {item["product_code"] for item in products.json()["data"]["items"]} == {
        "MED-BASIC",
        "MED-UPGRADE",
        "MED-SENIOR",
        "ACC-FAMILY",
    }

    medical = client.get("/api/v1/h5/products?category=MEDICAL", headers=headers)
    accident = client.get("/api/v1/h5/products?category=ACCIDENT", headers=headers)
    unknown = client.get("/api/v1/h5/products?category=UNKNOWN", headers=headers)
    assert medical.json()["data"]["total"] == 3
    assert accident.json()["data"]["total"] == 1
    assert unknown.json()["data"]["total"] == 0


def test_h5_validation_refresh_rotation_and_logout_use_real_mysql(
    client: TestClient,
) -> None:
    invalid_mobile = client.post(
        "/api/v1/h5/auth/login",
        json={"mobile": "1abcdefghij", "code": "147258"},
    )
    assert invalid_mobile.status_code == 400
    assert invalid_mobile.json()["error"]["code"] == "VALIDATION_ERROR"

    wrong_otp = client.post(
        "/api/v1/h5/auth/login",
        json={"mobile": "18899990001", "code": "000000"},
    )
    assert wrong_otp.status_code == 400
    assert wrong_otp.json()["error"]["code"] == "INVALID_OTP"

    tokens = h5_login(client)
    refreshed = client.post(
        "/api/v1/h5/auth/refresh",
        json={"refresh_token": tokens["refresh_token"]},
    )
    assert refreshed.status_code == 200
    refreshed_tokens = refreshed.json()["data"]["tokens"]

    reused = client.post(
        "/api/v1/h5/auth/refresh",
        json={"refresh_token": tokens["refresh_token"]},
    )
    assert reused.status_code == 401
    assert reused.json()["error"]["code"] == "AUTH_REQUIRED"

    headers = {"Authorization": f"Bearer {refreshed_tokens['access_token']}"}
    logout = client.post("/api/v1/h5/auth/logout", headers=headers)
    assert logout.status_code == 200
    after_logout = client.get("/api/v1/h5/me", headers=headers)
    assert after_logout.status_code == 401


@pytest.mark.parametrize(
    ("username", "role", "scope"),
    [
        ("admin", "SUPER_ADMIN", "ALL"),
        ("operator01", "OPERATOR", "MASKED_ALL"),
        ("reviewer01", "REVIEWER", "READ_ONLY"),
        ("sales_a", "SALESPERSON", "SELF"),
        ("sales_b", "SALESPERSON", "SELF"),
    ],
)
def test_admin_role_and_scope_matrix_uses_real_mysql(
    client: TestClient,
    username: str,
    role: str,
    scope: str,
) -> None:
    tokens = admin_login(client, username)
    profile = client.get(
        "/api/v1/admin/me",
        headers={"Authorization": f"Bearer {tokens['access_token']}"},
    )
    assert profile.status_code == 200
    assert profile.json()["data"]["username"] == username
    assert profile.json()["data"]["roles"] == [role]
    assert profile.json()["data"]["data_scope"] == scope


def test_admin_invalid_credentials_do_not_reveal_account_existence(
    client: TestClient,
) -> None:
    wrong_password = client.post(
        "/api/v1/admin/auth/login",
        json={"username": "admin", "password": "wrong-password"},
    )
    unknown_user = client.post(
        "/api/v1/admin/auth/login",
        json={"username": "not_exists", "password": "wrong-password"},
    )
    assert wrong_password.status_code == 401
    assert unknown_user.status_code == 401
    assert wrong_password.json()["error"] == unknown_user.json()["error"]


def test_h5_and_admin_tokens_cannot_cross_audiences(client: TestClient) -> None:
    h5_tokens = h5_login(client)
    admin_tokens = admin_login(client, "admin")
    h5_headers = {"Authorization": f"Bearer {h5_tokens['access_token']}"}
    admin_headers = {"Authorization": f"Bearer {admin_tokens['access_token']}"}

    assert client.get("/api/v1/admin/me", headers=h5_headers).status_code == 401
    assert client.get("/api/v1/h5/me", headers=admin_headers).status_code == 401
    assert client.get("/api/v1/h5/products", headers=admin_headers).status_code == 401


def test_request_id_and_cors_contract(client: TestClient) -> None:
    health = client.get(
        "/api/v1/system/health/live",
        headers={"X-Request-ID": "req_mysql_integration"},
    )
    assert health.headers["X-Request-ID"] == "req_mysql_integration"
    assert health.json()["meta"]["request_id"] == "req_mysql_integration"
    assert health.json()["meta"]["server_time"].endswith("Z")

    allowed = client.options(
        "/api/v1/h5/auth/login",
        headers={
            "Origin": "http://127.0.0.1:5173",
            "Access-Control-Request-Method": "POST",
        },
    )
    denied = client.options(
        "/api/v1/h5/auth/login",
        headers={
            "Origin": "http://untrusted.example",
            "Access-Control-Request-Method": "POST",
        },
    )
    assert allowed.headers["access-control-allow-origin"] == "http://127.0.0.1:5173"
    assert "access-control-allow-origin" not in denied.headers