harness_zbt/backend/tests/test_enrollment_flow.py

from datetime import UTC, datetime

from fastapi.testclient import TestClient

from zbt.commands.seed import build_seed_manifest
from zbt.core.config import Settings
from zbt.core.passwords import PasswordService
from zbt.domains.catalog.repository import InMemoryCatalogRepository
from zbt.domains.enrollment.repository import InMemoryEnrollmentRepository
from zbt.domains.identity.repository import InMemoryIdentityRepository
from zbt.main import create_app


def build_test_app():
    catalog = InMemoryCatalogRepository()
    identities = InMemoryIdentityRepository()
    manifest = build_seed_manifest(PasswordService().hash("zaq1XSW@"))
    for admin in manifest.admin_users:
        identities.save_admin_user(admin)
    for product in manifest.products:
        catalog.save_product(product)
    for version in manifest.versions:
        catalog.save_version(version)
    return create_app(
        settings=Settings(
            app_env="test",
            jwt_access_secret="a" * 32,
            jwt_refresh_secret="b" * 32,
            field_encryption_key="c" * 32,
        ),
        identity_repository=identities,
        catalog_repository=catalog,
        enrollment_repository=InMemoryEnrollmentRepository(),
        clock=lambda: datetime(2026, 7, 26, tzinfo=UTC),
    )


def login_headers(client: TestClient) -> dict[str, str]:
    token = client.post(
        "/api/v1/h5/auth/login",
        json={"mobile": "18800000001", "code": "147258"},
    ).json()["data"]["tokens"]["access_token"]
    return {"Authorization": f"Bearer {token}"}


def test_eligible_customer_receives_deterministic_quote() -> None:
    with TestClient(build_test_app()) as client:
        headers = login_headers(client)
        product = next(
            item
            for item in client.get("/api/v1/h5/products", headers=headers).json()["data"]["items"]
            if item["product_code"] == "MED-SENIOR"
        )
        response = client.post(
            "/api/v1/h5/quotes",
            headers=headers,
            json={
                "product_id": product["product_id"],
                "plan_id": product["plans"][0]["id"],
                "insured": {
                    "age": 65,
                    "region_code": "510100",
                    "occupation_code": "GENERAL",
                },
                "relationship": "PARENT",
            },
        )

    assert response.status_code == 201
    assert response.json()["data"]["eligible"] is True
    assert response.json()["data"]["premium_cents"] == 19900
    assert response.json()["data"]["rate_version"] == "rate-v1"


def test_confirmed_draft_creates_one_idempotent_order() -> None:
    with TestClient(build_test_app()) as client:
        headers = login_headers(client)
        product = next(
            item
            for item in client.get("/api/v1/h5/products", headers=headers).json()["data"]["items"]
            if item["product_code"] == "MED-SENIOR"
        )
        quote_id = client.post(
            "/api/v1/h5/quotes",
            headers=headers,
            json={
                "product_id": product["product_id"],
                "plan_id": product["plans"][0]["id"],
                "insured": {
                    "age": 65,
                    "region_code": "510100",
                    "occupation_code": "GENERAL",
                },
                "relationship": "PARENT",
            },
        ).json()["data"]["quote_id"]
        draft = client.post(
            "/api/v1/h5/enrollment-drafts",
            headers=headers,
            json={
                "quote_id": quote_id,
                "applicant": {"name": "张三", "id_no": "510100199001010001"},
                "insured": {"name": "张父", "id_no": "510100196101010001"},
                "contact": {"mobile": "18800000001"},
            },
        )
        confirmation = client.post(
            f"/api/v1/h5/enrollment-drafts/{draft.json()['data']['draft_id']}/confirmation",
            headers=headers,
        )
        order_headers = {
            **headers,
            "Idempotency-Key": "order-client-001",
        }
        first = client.post(
            "/api/v1/h5/orders",
            headers=order_headers,
            json={
                "draft_id": draft.json()["data"]["draft_id"],
                "confirmation_token": confirmation.json()["data"]["confirmation_token"],
            },
        )
        repeated = client.post(
            "/api/v1/h5/orders",
            headers=order_headers,
            json={
                "draft_id": draft.json()["data"]["draft_id"],
                "confirmation_token": confirmation.json()["data"]["confirmation_token"],
            },
        )

    assert draft.status_code == 201
    assert confirmation.status_code == 201
    assert first.status_code == 201
    assert first.json()["data"]["status"] == "PENDING_PAYMENT"
    assert first.json()["data"]["amount_cents"] == 19900
    assert repeated.status_code == 200
    assert repeated.json()["data"]["order_id"] == first.json()["data"]["order_id"]


def test_mock_payment_callback_is_idempotent_and_issues_one_policy() -> None:
    with TestClient(build_test_app()) as client:
        headers = login_headers(client)
        product = next(
            item
            for item in client.get("/api/v1/h5/products", headers=headers).json()["data"]["items"]
            if item["product_code"] == "MED-SENIOR"
        )
        quote_id = client.post(
            "/api/v1/h5/quotes",
            headers=headers,
            json={
                "product_id": product["product_id"],
                "plan_id": product["plans"][0]["id"],
                "insured": {
                    "age": 65,
                    "region_code": "510100",
                    "occupation_code": "GENERAL",
                },
                "relationship": "PARENT",
            },
        ).json()["data"]["quote_id"]
        draft_id = client.post(
            "/api/v1/h5/enrollment-drafts",
            headers=headers,
            json={
                "quote_id": quote_id,
                "applicant": {"name": "张三", "id_no": "510100199001010001"},
                "insured": {"name": "张父", "id_no": "510100196101010001"},
                "contact": {"mobile": "18800000001"},
            },
        ).json()["data"]["draft_id"]
        confirmation_token = client.post(
            f"/api/v1/h5/enrollment-drafts/{draft_id}/confirmation",
            headers=headers,
        ).json()["data"]["confirmation_token"]
        order = client.post(
            "/api/v1/h5/orders",
            headers={**headers, "Idempotency-Key": "order-client-002"},
            json={
                "draft_id": draft_id,
                "confirmation_token": confirmation_token,
            },
        ).json()["data"]
        payment = client.post(
            f"/api/v1/h5/orders/{order['order_id']}/payments",
            headers={**headers, "Idempotency-Key": "payment-client-001"},
        )
        invalid_callback = client.post(
            "/api/v1/callbacks/mock-payment",
            headers={"X-Mock-Pay-Signature": "invalid-signature"},
            json={
                "callback_no": "CB-INVALID-001",
                "provider_transaction_no": "MP-INVALID-001",
                "payment_no": payment.json()["data"]["payment_no"],
                "status": "SUCCEEDED",
                "amount_cents": 19900,
                "occurred_at": "2026-07-25T08:30:00.000Z",
            },
        )
        first_callback = client.post(
            f"/api/v1/dev/mock-payments/{payment.json()['data']['payment_id']}/complete",
            headers=headers,
        )
        repeated_callback = client.post(
            f"/api/v1/dev/mock-payments/{payment.json()['data']['payment_id']}/complete",
            headers=headers,
        )
        policies = client.get("/api/v1/h5/policies", headers=headers)
        admin_token = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        ).json()["data"]["tokens"]["access_token"]
        admin_dashboard = client.get(
            "/api/v1/admin/dashboard",
            headers={"Authorization": f"Bearer {admin_token}"},
        )
        admin_orders = client.get(
            "/api/v1/admin/orders",
            headers={"Authorization": f"Bearer {admin_token}"},
        )

    assert payment.status_code == 201
    assert invalid_callback.status_code == 401
    assert first_callback.status_code == 200
    assert first_callback.json()["data"]["payment_status"] == "SUCCEEDED"
    assert first_callback.json()["data"]["order_status"] == "ISSUED"
    assert first_callback.json()["data"]["policy"]["status"] == "ACTIVE"
    assert (
        repeated_callback.json()["data"]["policy"]["policy_id"]
        == (first_callback.json()["data"]["policy"]["policy_id"])
    )
    assert policies.json()["data"]["total"] == 1
    policy = policies.json()["data"]["items"][0]
    assert policy["product_name"] == "银龄守护医疗险"
    assert policy["plan_name"] == "银龄标准计划"
    assert policy["relationship"] == "PARENT"
    assert policy["order_no"] == order["order_no"]
    assert policy["applicant"]["name"] == "张三"
    assert policy["insured"]["name"] == "张父"
    assert admin_dashboard.status_code == 200
    assert admin_dashboard.json()["data"]["order_count"] == 1
    assert admin_dashboard.json()["data"]["policy_count"] == 1
    admin_order = admin_orders.json()["data"]["items"][0]
    assert admin_order["product_name"] == "银龄守护医疗险"
    assert admin_order["plan_name"] == "银龄标准计划"
    assert admin_order["relationship"] == "PARENT"
    assert admin_order["applicant"]["name"] == "张三"
    assert admin_order["contact"]["mobile"] == "18800000001"
    assert admin_order["payment"]["status"] == "SUCCEEDED"
    assert admin_order["policy_no"] == policy["policy_no"]


def test_super_admin_can_update_role_permissions() -> None:
    with TestClient(build_test_app()) as client:
        admin_token = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        ).json()["data"]["tokens"]["access_token"]
        headers = {"Authorization": f"Bearer {admin_token}"}

        roles = client.get("/api/v1/admin/roles", headers=headers)
        update = client.put(
            "/api/v1/admin/roles/OPERATOR",
            headers=headers,
            json={
                "permissions": [
                    "dashboard:read",
                    "order:read",
                    "policy:read",
                    "role:read",
                ],
                "data_scope": "MASKED_ALL",
            },
        )
        protected = client.put(
            "/api/v1/admin/roles/SUPER_ADMIN",
            headers=headers,
            json={"permissions": [], "data_scope": "SELF"},
        )
        operator_login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "operator01", "password": "zaq1XSW@"},
        )

    assert roles.status_code == 200
    assert len(roles.json()["data"]["items"]) == 4
    assert update.status_code == 200
    assert update.json()["data"]["permissions"] == [
        "dashboard:read",
        "order:read",
        "policy:read",
        "role:read",
    ]
    assert protected.status_code == 409
    assert operator_login.json()["data"]["user"]["permissions"] == [
        "dashboard:read",
        "order:read",
        "policy:read",
        "role:read",
    ]


def test_super_admin_can_maintain_admin_accounts_and_customers() -> None:
    with TestClient(build_test_app()) as client:
        customer_login = client.post(
            "/api/v1/h5/auth/login",
            json={"mobile": "18800000001", "code": "147258"},
        )
        customer_id = customer_login.json()["data"]["user"]["id"]
        admin_token = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        ).json()["data"]["tokens"]["access_token"]
        headers = {"Authorization": f"Bearer {admin_token}"}

        created = client.post(
            "/api/v1/admin/users/admins",
            headers=headers,
            json={
                "username": "service01",
                "password": "Initial1!",
                "display_name": "客服专员",
                "role_code": "OPERATOR",
            },
        )
        created_user = created.json()["data"]
        updated = client.put(
            f"/api/v1/admin/users/admins/{created_user['id']}",
            headers=headers,
            json={
                "display_name": "客服主管",
                "status": "ACTIVE",
                "role_code": "REVIEWER",
            },
        )
        password_reset = client.post(
            f"/api/v1/admin/users/admins/{created_user['id']}/password",
            headers=headers,
            json={"new_password": "Changed1!"},
        )
        customer_disabled = client.put(
            f"/api/v1/admin/users/customers/{customer_id}/status",
            headers=headers,
            json={"status": "DISABLED"},
        )
        users = client.get("/api/v1/admin/users", headers=headers)
        new_login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "service01", "password": "Changed1!"},
        )
        disabled_customer_login = client.post(
            "/api/v1/h5/auth/login",
            json={"mobile": "18800000001", "code": "147258"},
        )

    assert created.status_code == 201
    assert updated.status_code == 200
    assert updated.json()["data"]["display_name"] == "客服主管"
    assert updated.json()["data"]["roles"] == ["REVIEWER"]
    assert password_reset.status_code == 200
    assert customer_disabled.json()["data"]["status"] == "DISABLED"
    assert len(users.json()["data"]["admin_users"]) == 10
    assert len(users.json()["data"]["customers"]) == 1
    assert new_login.status_code == 200
    assert disabled_customer_login.status_code == 403