from __future__ import annotations import datetime from decimal import Decimal import pytest from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import StaticPool from backend.services.operations import ( get_model, get_monthly_monitoring_result, list_models, parse_monitor_month, ) from common.config import Settings from common.db.models.operations import ( OperationsBase, OpsBank, OpsModelInstance, OpsModelVersion, OpsMonitorBatch, OpsMonitorEvaluation, OpsMonitorResult, OpsMonitorReview, ) WORKSPACE_ID = "W" * 26 OTHER_WORKSPACE_ID = "X" * 26 MODEL_INSTANCE_ID = "M" * 26 MODEL_VERSION_ID = "V" * 26 MODEL_ID = "JC-STD-001" @pytest.fixture async def session(): engine = create_async_engine( "sqlite+aiosqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False}, ) async with engine.begin() as connection: await connection.run_sync(OperationsBase.metadata.create_all) factory = async_sessionmaker(engine, expire_on_commit=False) async with factory() as database_session: await _seed(database_session) await database_session.commit() yield database_session await engine.dispose() async def _seed(session) -> None: bank = OpsBank( bank_id="B" * 26, workspace_id=WORKSPACE_ID, bank_code="JC", bank_name="江城银行", is_wuji_bank=1, bank_status="active", source_bank_ref="bank-jc", ) version = OpsModelVersion( model_version_id=MODEL_VERSION_ID, workspace_id=WORKSPACE_ID, model_instance_id=MODEL_INSTANCE_ID, version_label="v2.3", version_status="active", last_iteration_date=datetime.date(2026, 6, 18), source_version_ref="version-jc-std-v23", ) model = OpsModelInstance( model_instance_id=MODEL_INSTANCE_ID, workspace_id=WORKSPACE_ID, bank_id=bank.bank_id, category_code="std", model_id=MODEL_ID, model_name="标准A卡", model_status="normal", current_version_id=version.model_version_id, common_model_name="标准A卡通用版 v2.3", source_model_ref="model-jc-std", ) old_batch = OpsMonitorBatch( batch_id="1" * 26, workspace_id=WORKSPACE_ID, source_batch_no="2026-07-r1", monitor_month=datetime.date(2026, 7, 1), revision_no=1, batch_status="published", published_at=datetime.datetime(2026, 7, 20, 8, 0), ) new_batch = OpsMonitorBatch( batch_id="2" * 26, workspace_id=WORKSPACE_ID, source_batch_no="2026-07-r2", monitor_month=datetime.date(2026, 7, 1), revision_no=2, batch_status="published", published_at=datetime.datetime(2026, 7, 21, 8, 0), ) old_result = OpsMonitorResult( monitor_result_id="3" * 26, workspace_id=WORKSPACE_ID, batch_id=old_batch.batch_id, model_instance_id=MODEL_INSTANCE_ID, model_version_id=MODEL_VERSION_ID, monitor_month=datetime.date(2026, 7, 1), ranking_result="matched", ks_value=Decimal("0.41000000"), psi_value=Decimal("0.09000000"), ) new_result = OpsMonitorResult( monitor_result_id="4" * 26, workspace_id=WORKSPACE_ID, batch_id=new_batch.batch_id, model_instance_id=MODEL_INSTANCE_ID, model_version_id=MODEL_VERSION_ID, monitor_month=datetime.date(2026, 7, 1), ranking_result="unmatched", ks_value=Decimal("0.36840000"), psi_value=Decimal("0.27400000"), ) evaluation = OpsMonitorEvaluation( evaluation_id="5" * 26, workspace_id=WORKSPACE_ID, monitor_result_id=new_result.monitor_result_id, rule_version_id="6" * 26, rule_item_id="7" * 26, ks_mom_drop_rate=Decimal("0.24100000"), secondary_level2_hits_6m=3, abnormal_level="level3", monitor_grade="C", reason_code="R-C-001", reason_text_snapshot="排序性不符、KS低、PSI高", action_snapshot="建议启动模型微调或重构评估", is_current=1, evaluated_at=datetime.datetime(2026, 7, 21, 9, 0), ) review = OpsMonitorReview( review_id="8" * 26, workspace_id=WORKSPACE_ID, monitor_result_id=new_result.monitor_result_id, evaluation_id=evaluation.evaluation_id, review_stage="model_initial", review_status="handled", decision_code="tune_or_rebuild", handling_note="建议评估重构", handled_by="U" * 26, handled_at=datetime.datetime(2026, 7, 22, 10, 30), ) other_bank = OpsBank( bank_id="C" * 26, workspace_id=OTHER_WORKSPACE_ID, bank_code="OTHER", bank_name="其他银行", bank_status="active", source_bank_ref="bank-other", ) other_model = OpsModelInstance( model_instance_id="N" * 26, workspace_id=OTHER_WORKSPACE_ID, bank_id=other_bank.bank_id, category_code="std", model_id="OTHER-STD-001", model_name="其他模型", model_status="normal", source_model_ref="model-other", ) session.add_all( [ bank, version, model, old_batch, new_batch, old_result, new_result, evaluation, review, other_bank, other_model, ] ) async def test_list_models_uses_latest_revision_and_workspace(session) -> None: models = await list_models(session, WORKSPACE_ID) assert len(models) == 1 model = models[0] assert model.model_id == MODEL_ID assert model.ks == 36.84 assert model.psi == 27.4 assert model.ks_mom_drop == 24.1 assert model.ranking_result == "不符" assert model.last_processed_at == "2026-07-22" assert model.is_wuji_bank is True async def test_list_models_filters_and_detail(session) -> None: assert len(await list_models(session, WORKSPACE_ID, bank="江城银行")) == 1 assert len(await list_models(session, WORKSPACE_ID, category="big")) == 0 assert len(await list_models(session, WORKSPACE_ID, keyword="v2.3")) == 1 model = await get_model(session, WORKSPACE_ID, MODEL_ID) assert model is not None assert model.model_version == "v2.3" assert model.common_model_name == "标准A卡通用版 v2.3" assert await get_model(session, OTHER_WORKSPACE_ID, MODEL_ID) is None async def test_get_monthly_monitoring_result(session) -> None: result = await get_monthly_monitoring_result( session, WORKSPACE_ID, MODEL_ID, datetime.date(2026, 7, 1), ) assert result is not None assert result.monitor_month == "2026-07" assert result.ks == 36.84 assert result.psi == 27.4 assert result.secondary_hits_6m == 3 def test_monitor_month_and_operations_url() -> None: assert parse_monitor_month("2026-07") == datetime.date(2026, 7, 1) with pytest.raises(ValueError): parse_monitor_month("2026-13") settings = Settings( database_url="mysql+asyncmy://user:pass@db:3306/model_platform?charset=utf8mb4", _env_file=None, ) assert settings.operations_database_url is not None assert "/model_operations?" in settings.operations_database_url