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 app.core.config import Settings
from app.core.passwords import PasswordService
from app.domains.catalog.models import Plan, Product, ProductVersion
from app.domains.catalog.repository import InMemoryCatalogRepository
from app.domains.identity.models import AdminUser, H5User
from app.infrastructure.mysql.core_models import CoreBase, PlanRecord
from app.infrastructure.mysql.repositories import (
    SqlAlchemyCatalogRepository,
    SqlAlchemyIdentityRepository,
)
from app.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"] == "系统管理员"