harness_zbt/backend/tests/test_attribution.py

from datetime import UTC, datetime

from fastapi.testclient import TestClient

from app.commands.seed import build_seed_manifest
from app.core.config import Settings
from app.core.passwords import PasswordService
from app.domains.agent.repository import InMemoryAgentRepository
from app.domains.attribution.models import (
    OrderAttribution,
    PromotionCode,
    Salesperson,
)
from app.domains.attribution.repository import InMemoryAttributionRepository
from app.domains.attribution.service import AttributionService
from app.domains.catalog.repository import InMemoryCatalogRepository
from app.domains.enrollment.models import EnrollmentOrder
from app.domains.enrollment.repository import InMemoryEnrollmentRepository
from app.domains.identity.models import AdminUser, H5User
from app.domains.identity.repository import InMemoryIdentityRepository
from app.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"