harness_zbt/backend/tests/test_attribution.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.agent.repository import InMemoryAgentRepository
from zbt.domains.attribution.models import (
    OrderAttribution,
    PromotionCode,
    Salesperson,
)
from zbt.domains.attribution.repository import InMemoryAttributionRepository
from zbt.domains.attribution.service import AttributionService
from zbt.domains.catalog.repository import InMemoryCatalogRepository
from zbt.domains.enrollment.models import EnrollmentOrder
from zbt.domains.enrollment.repository import InMemoryEnrollmentRepository
from zbt.domains.identity.models import AdminUser, H5User
from zbt.domains.identity.repository import InMemoryIdentityRepository
from zbt.main import create_app

NOW = datetime(2026, 7, 26, 9, tzinfo=UTC)


def _seed_attribution(
    repository: InMemoryAttributionRepository,
    *,
    admin_user_id: str = "01ADMIN0000000000000000001",
) -> tuple[Salesperson, PromotionCode]:
    salesperson = Salesperson(
        id="01SALES0000000000000000001",
        admin_user_id=admin_user_id,
        code="SALES-A",
        name="业务员A",
        status="ACTIVE",
        created_at=NOW,
    )
    code = PromotionCode(
        id="01PROMO0000000000000000001",
        code="ZBT-SALES-A",
        salesperson_id=salesperson.id,
        channel="GROUND_PROMOTION",
        status="ACTIVE",
        created_at=NOW,
    )
    repository.save_salesperson(salesperson)
    repository.save_promotion_code(code)
    return salesperson, code


def _order(order_id: str, user_id: str, amount_cents: int = 19900) -> EnrollmentOrder:
    return EnrollmentOrder(
        id=order_id,
        order_no=f"ORD-{order_id[-4:]}",
        user_id=user_id,
        quote_id=f"QUOTE-{order_id}"[-26:],
        draft_id=f"DRAFT-{order_id}"[-26:],
        confirmation_id=f"CONFIRM-{order_id}"[-26:],
        idempotency_key=f"IDEMPOTENCY-{order_id}",
        product_id="01PRODUCT00000000000000001",
        product_version_id="01VERSION00000000000000001",
        plan_id="01PLAN0000000000000000001",
        applicant_snapshot={
            "name": "周明远",
            "id_no": "510104198801011234",
        },
        insured_snapshot={
            "name": "周建国",
            "id_no": "510104196001011234",
        },
        amount_cents=amount_cents,
        currency="CNY",
        status="ISSUED",
        created_at=NOW,
    )


def test_first_touch_attribution_is_stable_and_order_snapshot_is_idempotent() -> None:
    repository = InMemoryAttributionRepository()
    salesperson, code = _seed_attribution(repository)
    service = AttributionService(repository, lambda: NOW)
    user = H5User(
        id="01H5USER000000000000000001",
        mobile="18800000001",
        mobile_masked="188****0001",
        display_name="体验用户",
        status="ACTIVE",
        created_at=NOW,
    )

    first = service.capture_visit(user, code.code)
    second = service.capture_visit(user, code.code)
    order = _order("01ORDER0000000000000000001", user.id)
    service.record_order(order)
    service.record_order(order)

    assert first["first_touch"] is True
    assert second["first_touch"] is False
    lead = repository.get_lead_by_user(user.id)
    assert lead is not None
    assert lead.salesperson_id == salesperson.id
    assert lead.visit_count == 2
    performance = service.performance(salesperson.admin_user_id)
    assert performance["visit_count"] == 2
    assert performance["lead_count"] == 1
    assert performance["order_count"] == 1
    assert performance["premium_cents"] == 19900
    assert service.order_ids(salesperson.admin_user_id) == {order.id}


def test_h5_referral_login_and_admin_self_scope_form_a_closed_loop() -> None:
    identities = InMemoryIdentityRepository()
    admin = AdminUser(
        id="01ADMIN0000000000000000001",
        username="sales_a",
        password_hash=PasswordService().hash("zaq1XSW@"),
        display_name="业务员A",
        status="ACTIVE",
        roles=("SALESPERSON",),
        permissions=(
            "dashboard:read",
            "order:read",
            "attribution:read",
            "product:read",
            "product:write",
        ),
        data_scope="SELF",
    )
    identities.save_admin_user(admin)
    attribution = InMemoryAttributionRepository()
    salesperson, code = _seed_attribution(attribution, admin_user_id=admin.id)
    catalog = InMemoryCatalogRepository()
    manifest = build_seed_manifest()
    for product in manifest.products:
        catalog.save_product(product)
    for version in manifest.versions:
        catalog.save_version(version)
    enrollment = InMemoryEnrollmentRepository()
    app = create_app(
        settings=Settings(app_env="test"),
        identity_repository=identities,
        catalog_repository=catalog,
        agent_repository=InMemoryAgentRepository(),
        enrollment_repository=enrollment,
        attribution_repository=attribution,
    )

    with TestClient(app) as client:
        h5_login = client.post(
            "/api/v1/h5/auth/login",
            json={
                "mobile": "18800000001",
                "code": "147258",
                "referral_code": code.code,
            },
        )
        assert h5_login.status_code == 200
        h5_user_id = h5_login.json()["data"]["user"]["id"]
        assert h5_login.json()["data"]["attribution"]["attributed"] is True
        owned_order = _order("01ORDER0000000000000000001", h5_user_id)
        other_order = _order("01ORDER0000000000000000002", "01OTHERUSER0000000000000001")
        enrollment.save_order(owned_order)
        enrollment.save_order(other_order)
        attribution.save_order_attribution(
            OrderAttribution(
                id="01ATTR00000000000000000001",
                order_id=owned_order.id,
                h5_user_id=h5_user_id,
                salesperson_id=salesperson.id,
                promotion_code_id=code.id,
                source_code=code.code,
                amount_cents=owned_order.amount_cents,
                attributed_at=NOW,
            )
        )

        admin_login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "sales_a", "password": "zaq1XSW@"},
        )
        headers = {
            "Authorization": (f"Bearer {admin_login.json()['data']['tokens']['access_token']}")
        }
        orders = client.get("/api/v1/admin/orders", headers=headers)
        overview = client.get(
            "/api/v1/admin/attribution/overview",
            headers=headers,
        )
        products = client.get("/api/v1/admin/products", headers=headers)
        product_id = products.json()["data"]["items"][0]["product_id"]
        disabled = client.put(
            f"/api/v1/admin/products/{product_id}/status",
            headers=headers,
            json={"status": "INACTIVE"},
        )

    assert orders.status_code == 200
    assert orders.json()["data"]["total"] == 1
    visible_order = orders.json()["data"]["items"][0]
    assert visible_order["order_id"] == owned_order.id
    assert visible_order["applicant"]["id_no"] == "510104********1234"
    assert overview.json()["data"]["salesperson_count"] == 1
    assert overview.json()["data"]["order_count"] == 1
    assert disabled.status_code == 200
    assert disabled.json()["data"]["status"] == "INACTIVE"