harness_zbt/backend/tests/integration/test_sqlalchemy_repositories.py

from dataclasses import replace
from datetime import UTC, datetime

from fastapi.testclient import TestClient
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from zbt.core.config import Settings
from zbt.core.passwords import PasswordService
from zbt.domains.catalog.models import Plan, Product, ProductVersion
from zbt.domains.catalog.repository import InMemoryCatalogRepository
from zbt.domains.identity.models import AdminUser, H5User
from zbt.infrastructure.mysql.core_models import CoreBase, PlanRecord
from zbt.infrastructure.mysql.repositories import (
    SqlAlchemyCatalogRepository,
    SqlAlchemyIdentityRepository,
)
from zbt.main import create_app


def test_sqlalchemy_repositories_round_trip_identity_and_catalog() -> None:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    CoreBase.metadata.create_all(engine)
    session_factory = sessionmaker(engine, class_=Session, expire_on_commit=False)
    identities = SqlAlchemyIdentityRepository(session_factory)
    catalog = SqlAlchemyCatalogRepository(session_factory)

    identities.save_h5_user(
        H5User(
            id="01USER00000000000000000001",
            mobile="18800000001",
            mobile_masked="188****0001",
            display_name="测试用户甲",
            status="ACTIVE",
            created_at=datetime(2026, 7, 25, tzinfo=UTC),
        )
    )
    catalog.save_product(
        Product(
            id="01PRODUCT00000000000000001",
            product_code="MED-BASIC",
            name="蓉惠基础医疗险",
            category="MEDICAL",
            summary="基础医疗保障",
            status="ACTIVE",
        )
    )
    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(version)
    catalog.save_version(version)

    assert identities.get_h5_user_by_mobile("18800000001").display_name == "测试用户甲"
    assert catalog.list_products()[0].product_code == "MED-BASIC"
    loaded_version = catalog.list_versions("01PRODUCT00000000000000001")[0]
    assert loaded_version.version_no == "2.0.0"
    assert loaded_version.is_available_at(datetime(2026, 7, 25, tzinfo=UTC))
    with session_factory() as session:
        assert session.scalar(select(func.count()).select_from(PlanRecord)) == 1


def test_admin_session_loaded_from_sql_repository_remains_active() -> None:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    CoreBase.metadata.create_all(engine)
    session_factory = sessionmaker(engine, class_=Session, expire_on_commit=False)
    identities = SqlAlchemyIdentityRepository(session_factory)
    admin_user = AdminUser(
        id="01ADMIN0000000000000000001",
        username="admin",
        password_hash=PasswordService().hash("旧密码仅用于覆盖测试"),
        display_name="系统管理员",
        status="ACTIVE",
        roles=("SUPER_ADMIN",),
        permissions=("*",),
        data_scope="ALL",
    )
    identities.save_admin_user(admin_user)
    identities.save_admin_user(
        replace(
            admin_user,
            password_hash=PasswordService().hash("zaq1XSW@"),
        )
    )
    settings = Settings(
        app_env="test",
        jwt_access_secret="a" * 32,
        jwt_refresh_secret="b" * 32,
        field_encryption_key="c" * 32,
    )
    application = create_app(
        settings=settings,
        identity_repository=identities,
        catalog_repository=InMemoryCatalogRepository(),
    )

    with TestClient(application) as client:
        login = client.post(
            "/api/v1/admin/auth/login",
            json={"username": "admin", "password": "zaq1XSW@"},
        )
        token = login.json()["data"]["tokens"]["access_token"]
        profile = client.get(
            "/api/v1/admin/me",
            headers={"Authorization": f"Bearer {token}"},
        )

    assert profile.status_code == 200
    assert profile.json()["data"]["display_name"] == "系统管理员"