from datetime import UTC, datetime, timedelta import pytest from fastapi.testclient import TestClient from zbt.core.config import Settings from zbt.core.errors import AppError from zbt.domains.catalog.models import Plan, Product, ProductVersion from zbt.domains.catalog.repository import InMemoryCatalogRepository from zbt.domains.catalog.service import ProductCatalogService from zbt.domains.identity.repository import InMemoryIdentityRepository from zbt.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" )