""" StatementService 主绑定注入测试 """ from collections import Counter, defaultdict from services.file_service import FileService from services.statement_service import StatementService from services.statement_rule_samples import ( DEFAULT_LARGE_TRANSACTION_THRESHOLDS, build_large_transaction_seed_statements, ) def test_generate_statements_should_include_seeded_samples_before_noise(): """生成流水时必须先混入固定命中样本，而不是纯随机噪声。""" service = StatementService() statements = service._generate_statements(group_id=1000, log_id=20001, count=30) assert len(statements) >= 30 assert any(item["userMemo"] == "购买房产首付款" for item in statements) def test_large_transaction_seed_should_cover_all_eight_rules(): """大额交易样本生成器必须覆盖 8 条已实现规则的关键口径。""" statements = build_large_transaction_seed_statements(group_id=1000, log_id=20001) assert any( item["userMemo"] == "购买房产首付款" and item["drAmount"] > 0 for item in statements ) assert any( "税" in item["userMemo"] and item["drAmount"] > 0 for item in statements ) assert any( item["crAmount"] > DEFAULT_LARGE_TRANSACTION_THRESHOLDS["SINGLE_TRANSACTION_AMOUNT"] for item in statements ) assert sum( 1 for item in statements if item["customerName"] == "浙江远望贸易有限公司" and item["crAmount"] > 0 ) >= 3 assert sum( 1 for item in statements if item["cashType"] == "现金存款" and item["crAmount"] > DEFAULT_LARGE_TRANSACTION_THRESHOLDS["LARGE_CASH_DEPOSIT"] ) >= 1 assert any( item["userMemo"] == "手机银行转账" and item["drAmount"] > DEFAULT_LARGE_TRANSACTION_THRESHOLDS["FREQUENT_TRANSFER"] for item in statements ) same_day_cash_deposits = [ item for item in statements if item["cretNo"] == "330101198801010011" and item["trxDate"].startswith("2026-03-10") and item["crAmount"] > DEFAULT_LARGE_TRANSACTION_THRESHOLDS["LARGE_CASH_DEPOSIT"] ] assert len(same_day_cash_deposits) >= ( DEFAULT_LARGE_TRANSACTION_THRESHOLDS["FREQUENT_CASH_DEPOSIT"] + 1 ) def test_large_transaction_seed_income_should_avoid_salary_exclusion(): """收入样本不得误带工资代发关键词，否则会被后端过滤。""" statements = build_large_transaction_seed_statements(group_id=1000, log_id=20001) income_samples = [item for item in statements if item["crAmount"] > 0] assert income_samples assert all(item["customerName"] != "浙江兰溪农村商业银行股份有限公司" for item in income_samples) assert all( keyword not in item["userMemo"] for item in income_samples for keyword in ("代发", "工资", "奖金", "薪酬", "薪金") ) def test_generate_statements_should_fill_noise_up_to_requested_count(): """样本不足请求总数时，服务层需要自动补齐噪声流水。""" service = StatementService() statements = service._generate_statements(group_id=1000, log_id=20001, count=80) assert len(statements) == 80 def test_generate_statements_should_only_use_recognizable_identity_cards(): """命中样本和随机噪声都只能使用现库可识别的身份证号。""" service = StatementService() statements = service._generate_statements(group_id=1000, log_id=20005, count=1600) assert {item["cretNo"] for item in statements}.issubset( { "330101198801010011", "330101199001010022", "330101198802020033", "330101199202020044", } ) def test_get_bank_statement_should_keep_same_cached_result_for_same_log_id(): """同一 logId 首次生成后应复用缓存，避免分页结果漂移。""" service = StatementService() page1 = service.get_bank_statement( {"groupId": 1000, "logId": 30001, "pageNow": 1, "pageSize": 20} ) page2 = service.get_bank_statement( {"groupId": 1000, "logId": 30001, "pageNow": 1, "pageSize": 20} ) assert page1["data"]["bankStatementList"] == page2["data"]["bankStatementList"] def test_get_bank_statement_uses_primary_binding_from_file_service(monkeypatch): """同一 logId 的流水记录必须复用 FileService 中的主体与账号绑定。""" file_service = FileService() statement_service = StatementService(file_service=file_service) monkeypatch.setattr( file_service, "_generate_primary_binding", lambda: ("绑定主体", "6222000011112222"), ) response = file_service.fetch_inner_flow( { "groupId": 1001, "customerNo": "customer_001", "dataChannelCode": "test", "requestDateId": 20240101, "dataStartDateId": 20240101, "dataEndDateId": 20240131, "uploadUserId": 902001, } ) log_id = response["data"][0] record = file_service.file_records[log_id] statement_response = statement_service.get_bank_statement( { "groupId": 1001, "logId": log_id, "pageNow": 1, "pageSize": 5, } ) statements = statement_response["data"]["bankStatementList"] assert statements assert all(item["leName"] == record.primary_enterprise_name for item in statements) assert all(item["accountMaskNo"] == record.primary_account_no for item in statements) def test_get_bank_statement_contains_large_transaction_hit_samples(monkeypatch): """流水 Mock 首次生成时必须稳定包含可命中大额交易规则的样本簇。""" file_service = FileService() statement_service = StatementService(file_service=file_service) monkeypatch.setattr( file_service, "_generate_primary_binding", lambda: ("命中主体", "6222000099998888"), ) response = file_service.fetch_inner_flow( { "groupId": 1001, "customerNo": "customer_large_transaction", "dataChannelCode": "test", "requestDateId": 20240101, "dataStartDateId": 20240101, "dataEndDateId": 20240131, "uploadUserId": 902001, } ) log_id = response["data"][0] statement_response = statement_service.get_bank_statement( { "groupId": 1001, "logId": log_id, "pageNow": 1, "pageSize": 2000, } ) statements = statement_response["data"]["bankStatementList"] assert statements assert any( item["cretNo"] in { "330101198801010011", "330101199001010022", "330101198802020033", "330101199202020044", } for item in statements ) assert any("房产首付款" in item["userMemo"] for item in statements) assert any("税" in item["userMemo"] or "税务" in item["customerName"] for item in statements) income_amounts = defaultdict(float) cash_deposit_daily_counter = Counter() has_large_transfer = False for item in statements: if ( item["cretNo"] == "330101198802020033" and item["customerName"] == "浙江远望贸易有限公司" and item["crAmount"] > 0 ): income_amounts[(item["cretNo"], item["customerName"])] += item["crAmount"] if item["crAmount"] > 2000001 and "现金" in item["cashType"]: cash_deposit_daily_counter[(item["cretNo"], item["trxDate"][:10])] += 1 if item["drAmount"] > 100001 and item["userMemo"] == "手机银行转账": has_large_transfer = True assert any(amount > 50000001 for amount in income_amounts.values()) assert any(count >= 6 for count in cash_deposit_daily_counter.values()) assert has_large_transfer