102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
"""
|
|
第二期数据库基线服务测试
|
|
"""
|
|
|
|
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)
|