harness_zbt/backend/tests/test_product_catalog.py

from datetime import UTC, datetime, timedelta

import pytest
from fastapi.testclient import TestClient

from app.core.config import Settings
from app.core.errors import AppError
from app.domains.catalog.models import Plan, Product, ProductVersion
from app.domains.catalog.repository import InMemoryCatalogRepository
from app.domains.catalog.service import ProductCatalogService
from app.domains.identity.repository import InMemoryIdentityRepository
from app.main import create_app


def test_h5_catalog_only_returns_current_published_product_versions() -> None:
    now = datetime(2026, 7, 25, tzinfo=UTC)
    catalog = InMemoryCatalogRepository()
    catalog.save_product(
        Product(
            id="01PRODUCT00000000000000001",
            product_code="MED-BASIC",
            name="蓉惠基础医疗险",
            category="MEDICAL",
            summary="基础医疗保障",
            status="ACTIVE",
        )
    )
    catalog.save_version(
        ProductVersion(
            id="01VERSION00000000000000001",
            product_id="01PRODUCT00000000000000001",
            version_no="1.0.0",
            status="EXPIRED",
            effective_from=datetime(2024, 1, 1, tzinfo=UTC),
            effective_to=datetime(2025, 12, 31, tzinfo=UTC),
            plans=(Plan(id="01PLAN0000000000000000001", code="BASIC", name="基础计划"),),
        )
    )
    catalog.save_version(
        ProductVersion(
            id="01VERSION00000000000000002",
            product_id="01PRODUCT00000000000000001",
            version_no="2.0.0",
            status="PUBLISHED",
            effective_from=datetime(2026, 1, 1, tzinfo=UTC),
            effective_to=None,
            plans=(Plan(id="01PLAN0000000000000000002", code="BASIC", name="基础计划"),),
        )
    )
    catalog.save_version(
        ProductVersion(
            id="01VERSION00000000000000003",
            product_id="01PRODUCT00000000000000001",
            version_no="3.0.0",
            status="DRAFT",
            effective_from=datetime(2027, 1, 1, tzinfo=UTC),
            effective_to=None,
            plans=(Plan(id="01PLAN0000000000000000003", code="BASIC", name="基础计划"),),
        )
    )
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    app = create_app(
        settings=settings,
        identity_repository=InMemoryIdentityRepository(),
        catalog_repository=catalog,
        clock=lambda: now,
    )

    with TestClient(app) as client:
        token = client.post(
            "/api/v1/h5/auth/login",
            json={"mobile": "18800000001", "code": "147258"},
        ).json()["data"]["tokens"]["access_token"]
        response = client.get(
            "/api/v1/h5/products",
            headers={"Authorization": f"Bearer {token}"},
        )

    assert response.status_code == 200
    assert [item["version_no"] for item in response.json()["data"]["items"]] == ["2.0.0"]


def test_product_version_draft_publish_and_audit_lifecycle() -> None:
    now = datetime(2026, 7, 26, tzinfo=UTC)
    repository = InMemoryCatalogRepository()
    product = Product(
        id="01PRODUCT00000000000000001",
        product_code="MED-BASIC",
        name="智保基础医疗险",
        category="MEDICAL",
        summary="基础保障",
        status="ACTIVE",
    )
    source = ProductVersion(
        id="01VERSION00000000000000001",
        product_id=product.id,
        version_no="2.0.0",
        status="PUBLISHED",
        effective_from=datetime(2026, 1, 1, tzinfo=UTC),
        effective_to=None,
        terms_summary="原版本投保须知",
        plans=(
            Plan(
                id="01PLAN0000000000000000001",
                code="BASIC",
                name="基础计划",
                premium_cents=23900,
                coverage_amount_cents=200_000_000,
                min_age=0,
                max_age=65,
            ),
        ),
    )
    repository.save_product(product)
    repository.save_version(source)
    service = ProductCatalogService(repository, lambda: now)

    created = service.create_draft_version(
        product.id,
        version_no="3.0.0",
        effective_from=now + timedelta(days=30),
        source_version_id=source.id,
        actor_id="ADMIN-1",
        actor_name="管理员",
    )
    updated = service.update_draft_version(
        product.id,
        str(created["id"]),
        effective_from=now + timedelta(days=30),
        effective_to=None,
        rule_version="eligibility-v2",
        rate_version="rate-v2",
        terms_summary="新版投保须知",
        plans=[
            {
                **created["plans"][0],
                "premium_cents": 25900,
                "coverage_amount_cents": 300_000_000,
            }
        ],
        actor_id="ADMIN-1",
        actor_name="管理员",
    )
    published = service.publish_version(
        product.id,
        str(created["id"]),
        actor_id="ADMIN-1",
        actor_name="管理员",
    )

    versions = repository.list_versions(product.id)
    previous = next(version for version in versions if version.id == source.id)
    assert updated["plans"][0]["premium_cents"] == 25900
    assert published["status"] == "PUBLISHED"
    assert previous.effective_to == now + timedelta(days=30) - timedelta(microseconds=1)
    assert [log.action for log in repository.list_change_logs(product.id)] == [
        "VERSION_DRAFT_CREATED",
        "VERSION_DRAFT_UPDATED",
        "VERSION_PUBLISHED",
    ]

    with pytest.raises(AppError, match="只有草稿版本可以编辑"):
        service.update_draft_version(
            product.id,
            str(created["id"]),
            effective_from=now,
            effective_to=None,
            rule_version="eligibility-v3",
            rate_version="rate-v3",
            terms_summary="不可直接修改",
            plans=[],
            actor_id="ADMIN-1",
            actor_name="管理员",
        )


def test_create_product_starts_inactive_with_an_editable_initial_version() -> None:
    now = datetime(2026, 7, 26, tzinfo=UTC)
    repository = InMemoryCatalogRepository()
    service = ProductCatalogService(repository, lambda: now)

    created = service.create_product(
        product_code="health-new",
        name="智享健康医疗险",
        category="MEDICAL",
        summary="面向家庭客户的综合医疗保障。",
        version_no="1.0.0",
        effective_from=now + timedelta(days=1),
        rule_version="eligibility-v1",
        rate_version="rate-v1",
        terms_summary="投保前请阅读责任免除和等待期说明。",
        plans=[
            {
                "code": "STANDARD",
                "name": "标准计划",
                "summary": "覆盖住院医疗费用。",
                "status": "ACTIVE",
                "premium_cents": 29900,
                "coverage_amount_cents": 2_000_000_00,
                "min_age": 0,
                "max_age": 65,
            }
        ],
        actor_id="ADMIN-1",
        actor_name="管理员",
    )

    assert created["product"]["product_code"] == "HEALTH-NEW"
    assert created["product"]["status"] == "INACTIVE"
    assert created["initial_version"]["status"] == "DRAFT"
    assert created["initial_version"]["plans"][0]["premium_cents"] == 29900
    assert repository.list_change_logs(created["product"]["product_id"])[0].action == (
        "PRODUCT_CREATED"
    )