155 lines
4.4 KiB
Python
155 lines
4.4 KiB
Python
import pytest
|
|
|
|
from services.abnormal_account_baseline_service import AbnormalAccountBaselineService
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, connection):
|
|
self.connection = connection
|
|
|
|
def execute(self, sql, params=None):
|
|
self.connection.executed_sql.append(
|
|
{
|
|
"sql": sql,
|
|
"params": params,
|
|
}
|
|
)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self):
|
|
self.executed_sql = []
|
|
self.commit_count = 0
|
|
self.rollback_count = 0
|
|
|
|
def cursor(self):
|
|
return FakeCursor(self)
|
|
|
|
def commit(self):
|
|
self.commit_count += 1
|
|
|
|
def rollback(self):
|
|
self.rollback_count += 1
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
|
|
def test_apply_should_skip_when_abnormal_accounts_is_empty():
|
|
service = AbnormalAccountBaselineService()
|
|
fake_connection = FakeConnection()
|
|
service._connect = lambda: fake_connection
|
|
|
|
service.apply("330101199001010001", [])
|
|
|
|
assert fake_connection.executed_sql == []
|
|
assert fake_connection.commit_count == 0
|
|
assert fake_connection.rollback_count == 0
|
|
|
|
|
|
def test_apply_should_raise_when_fact_owner_mismatches_staff():
|
|
service = AbnormalAccountBaselineService()
|
|
|
|
with pytest.raises(RuntimeError, match="owner_id_card"):
|
|
service.apply(
|
|
"330101199001010001",
|
|
[
|
|
{
|
|
"account_no": "6222000000000001",
|
|
"owner_id_card": "330101199001010099",
|
|
"account_name": "测试员工工资卡",
|
|
"status": 2,
|
|
"effective_date": "2024-01-01",
|
|
"invalid_date": "2026-03-20",
|
|
"rule_code": "SUDDEN_ACCOUNT_CLOSURE",
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
def test_apply_should_insert_new_account_fact_by_account_no():
|
|
service = AbnormalAccountBaselineService()
|
|
fake_connection = FakeConnection()
|
|
service._connect = lambda: fake_connection
|
|
|
|
service.apply(
|
|
"330101199001010001",
|
|
[
|
|
{
|
|
"account_no": "6222000000000001",
|
|
"owner_id_card": "330101199001010001",
|
|
"account_name": "测试员工工资卡",
|
|
"status": 2,
|
|
"effective_date": "2024-01-01",
|
|
"invalid_date": "2026-03-20",
|
|
"rule_code": "SUDDEN_ACCOUNT_CLOSURE",
|
|
}
|
|
],
|
|
)
|
|
|
|
assert len(fake_connection.executed_sql) == 1
|
|
executed = fake_connection.executed_sql[0]
|
|
assert "INSERT INTO ccdi_account_info" in executed["sql"]
|
|
assert "create_by" in executed["sql"]
|
|
assert "update_by" in executed["sql"]
|
|
assert "created_by" not in executed["sql"]
|
|
assert "updated_by" not in executed["sql"]
|
|
assert executed["params"] == (
|
|
"6222000000000001",
|
|
"DEBIT",
|
|
"测试员工工资卡",
|
|
"EMPLOYEE",
|
|
"330101199001010001",
|
|
"兰溪农商银行",
|
|
"LXNCSY",
|
|
"CNY",
|
|
1,
|
|
"HIGH",
|
|
2,
|
|
"2024-01-01",
|
|
"2026-03-20",
|
|
"lsfx-mock-server",
|
|
"lsfx-mock-server",
|
|
)
|
|
assert fake_connection.commit_count == 1
|
|
assert fake_connection.rollback_count == 0
|
|
|
|
|
|
def test_apply_should_update_existing_account_fact_by_account_no():
|
|
service = AbnormalAccountBaselineService()
|
|
fake_connection = FakeConnection()
|
|
service._connect = lambda: fake_connection
|
|
|
|
service.apply(
|
|
"330101199001010001",
|
|
[
|
|
{
|
|
"account_no": "6222000000000001",
|
|
"owner_id_card": "330101199001010001",
|
|
"account_name": "测试员工结算卡",
|
|
"status": 1,
|
|
"effective_date": "2025-01-01",
|
|
"invalid_date": None,
|
|
"rule_code": "DORMANT_ACCOUNT_LARGE_ACTIVATION",
|
|
}
|
|
],
|
|
)
|
|
|
|
assert len(fake_connection.executed_sql) == 1
|
|
executed = fake_connection.executed_sql[0]
|
|
assert "ON DUPLICATE KEY UPDATE" in executed["sql"]
|
|
assert "update_by = VALUES(update_by)" in executed["sql"]
|
|
assert executed["params"][0] == "6222000000000001"
|
|
assert executed["params"][2] == "测试员工结算卡"
|
|
assert executed["params"][10] == 1
|
|
assert executed["params"][11] == "2025-01-01"
|
|
assert executed["params"][12] is None
|
|
assert fake_connection.commit_count == 1
|
|
assert fake_connection.rollback_count == 0
|