完成LSFX Mock第二期稳定随机命中后端实施

lsfx-mock-server/tests/test_phase2_baseline_service.py

@@ -0,0 +1,101 @@
+"""
+第二期数据库基线服务测试
+"""
+
+from contextlib import nullcontext
+
+from services.phase2_baseline_service import Phase2BaselineService
+
+
+def test_build_sql_plan_should_return_idempotent_sql_plan_for_selected_phase2_baselines():
+    """抽中第二期基线规则时，应生成幂等 SQL 计划。"""
+    service = Phase2BaselineService()
+
+    sql_plan = service.build_sql_plan(
+        staff_id_card="330101198801010011",
+        family_id_cards=["330101199001010022"],
+        baseline_rule_codes=[
+            "SUPPLIER_CONCENTRATION",
+            "HOUSE_REGISTRATION_MISMATCH",
+        ],
+    )
+
+    assert any("LSFXMOCKP2PUR001" in sql for sql in sql_plan)
+    assert any("LSFX Mock P2 HOUSE" in sql for sql in sql_plan)
+    assert any("'房产'" in sql for sql in sql_plan)
+    assert any("'正常'" in sql for sql in sql_plan)
+    assert any(sql.strip().startswith("DELETE") for sql in sql_plan)
+    assert any(sql.strip().startswith("INSERT") for sql in sql_plan)
+
+
+def test_build_sql_plan_should_skip_unselected_phase2_rules():
+    """未选中的规则不应写入无关 SQL。"""
+    service = Phase2BaselineService()
+
+    sql_plan = service.build_sql_plan(
+        staff_id_card="330101198801010011",
+        family_id_cards=[],
+        baseline_rule_codes=["SUPPLIER_CONCENTRATION"],
+    )
+
+    assert any("LSFXMOCKP2PUR001" in sql for sql in sql_plan)
+    assert not any("LSFX Mock P2 HOUSE" in sql for sql in sql_plan)
+    assert not any("ccdi_asset_info" in sql for sql in sql_plan)
+
+
+def test_build_sql_plan_should_use_staff_scope_for_family_asset_baselines():
+    """亲属资产基线应保留员工归属与亲属实际持有人的双字段语义。"""
+    service = Phase2BaselineService()
+
+    sql_plan = service.build_sql_plan(
+        staff_id_card="330101198801010011",
+        family_id_cards=["330101199001010022"],
+        baseline_rule_codes=["PROPERTY_FEE_REGISTRATION_MISMATCH"],
+    )
+
+    assert any("'330101198801010011'" in sql for sql in sql_plan)
+    assert any("'330101199001010022'" in sql for sql in sql_plan)
+    assert not any("'REAL_ESTATE'" in sql for sql in sql_plan)
+
+
+def test_apply_should_execute_generated_sql_plan(monkeypatch):
+    """apply() 应执行生成出的 SQL 计划，而不是只返回字符串。"""
+    service = Phase2BaselineService()
+    executed_sql = []
+    committed = {"value": False}
+
+    class FakeCursor:
+        def execute(self, sql):
+            executed_sql.append(sql.strip())
+
+    class FakeConnection:
+        def __init__(self):
+            self.cursor_instance = FakeCursor()
+
+        def cursor(self):
+            return nullcontext(self.cursor_instance)
+
+        def commit(self):
+            committed["value"] = True
+
+        def rollback(self):
+            raise AssertionError("测试路径不应触发回滚")
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+    monkeypatch.setattr(service, "_connect", lambda: FakeConnection())
+
+    result = service.apply(
+        staff_id_card="330101198801010011",
+        family_id_cards=["330101199001010022"],
+        baseline_rule_codes=["SUPPLIER_CONCENTRATION"],
+    )
+
+    assert result is None
+    assert committed["value"] is True
+    assert any("DELETE FROM ccdi_purchase_transaction" in sql for sql in executed_sql)
+    assert any("INSERT INTO ccdi_purchase_transaction" in sql for sql in executed_sql)