745 lines
28 KiB
Python
745 lines
28 KiB
Python
"""
|
||
收入完整性检测算法测试用例
|
||
|
||
覆盖场景：
|
||
1. 正常场景：充值与申报一致
|
||
2. 少报收入场景：存在隐瞒收入
|
||
3. 多报收入场景：申报收入超过充值
|
||
4. 无分成协议场景
|
||
5. 边界值测试
|
||
6. 异常情况处理
|
||
"""
|
||
import pytest
|
||
from datetime import datetime, date, timedelta
|
||
from decimal import Decimal
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select
|
||
|
||
from app.services.risk_detection.algorithms.revenue_integrity import RevenueIntegrityAlgorithm
|
||
from app.services.risk_detection.algorithms.base import DetectionContext
|
||
from app.models.risk_detection import RiskLevel
|
||
from app.models.streamer import StreamerInfo, PlatformRecharge
|
||
from app.models.tax_declaration import TaxDeclaration
|
||
|
||
|
||
class TestRevenueIntegrityAlgorithm:
|
||
"""收入完整性检测算法测试类"""
|
||
|
||
@pytest.fixture
|
||
def algorithm(self):
|
||
"""创建算法实例"""
|
||
return RevenueIntegrityAlgorithm()
|
||
|
||
@pytest.fixture
|
||
def mock_db_session(self):
|
||
"""创建Mock数据库会话"""
|
||
session = AsyncMock(spec=AsyncSession)
|
||
|
||
# Mock执行方法
|
||
session.execute = AsyncMock()
|
||
|
||
return session
|
||
|
||
@pytest.fixture
|
||
def streamer_info(self):
|
||
"""主播信息Mock数据"""
|
||
return {
|
||
"streamer_id": "ZB_TEST_001",
|
||
"streamer_name": "测试主播",
|
||
"entity_type": "individual",
|
||
"tax_registration_no": "TAX123456789",
|
||
"unified_social_credit_code": None,
|
||
"id_card_no": "110101199001011234",
|
||
}
|
||
|
||
@pytest.fixture
|
||
def mock_recharge_data(self):
|
||
"""Mock充值数据"""
|
||
return [
|
||
{
|
||
"recharge_id": "RC001",
|
||
"user_name": "测试用户",
|
||
"amount": 10000.0,
|
||
"time": datetime(2024, 1, 15, 10, 30, 0),
|
||
"payment_method": "bank_transfer",
|
||
},
|
||
{
|
||
"recharge_id": "RC002",
|
||
"user_name": "测试用户",
|
||
"amount": 20000.0,
|
||
"time": datetime(2024, 1, 20, 14, 20, 0),
|
||
"payment_method": "bank_transfer",
|
||
},
|
||
{
|
||
"recharge_id": "RC003",
|
||
"user_name": "测试用户",
|
||
"amount": 15000.0,
|
||
"time": datetime(2024, 1, 25, 9, 15, 0),
|
||
"payment_method": "bank_transfer",
|
||
},
|
||
]
|
||
|
||
@pytest.fixture
|
||
def mock_declaration_data(self):
|
||
"""Mock税务申报数据"""
|
||
return [
|
||
{
|
||
"declaration_id": "TAX001",
|
||
"taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01",
|
||
"sales_revenue": 45000.0,
|
||
"declaration_date": date(2024, 2, 15),
|
||
},
|
||
]
|
||
|
||
@pytest.fixture
|
||
def mock_contract_data(self):
|
||
"""Mock分成协议数据"""
|
||
return {
|
||
"contract_id": "CT001",
|
||
"streamer_ratio": 70.0, # 主播分成70%
|
||
"platform_ratio": 30.0, # 平台分成30%
|
||
"contract_start_date": date(2024, 1, 1),
|
||
"contract_end_date": date(2024, 12, 31),
|
||
}
|
||
|
||
|
||
class TestRevenueIntegrityNormal(TestRevenueIntegrityAlgorithm):
|
||
"""正常场景测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_revenue_match_with_contract(self, algorithm, mock_db_session, streamer_info,
|
||
mock_recharge_data, mock_declaration_data, mock_contract_data):
|
||
"""测试场景1：收入一致（有分成协议）"""
|
||
# 设置：充值总额45000元，分成70%，申报45000元
|
||
|
||
# Mock数据库查询结果
|
||
mock_db_session.execute.side_effect = [
|
||
# 查询主播信息
|
||
self._create_streamer_result(streamer_info),
|
||
# 查询充值数据
|
||
self._create_recharge_result(mock_recharge_data),
|
||
# 查询申报数据
|
||
self._create_declaration_result(mock_declaration_data),
|
||
# 查询分成协议
|
||
self._create_contract_result(mock_contract_data),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言
|
||
assert result.risk_level == RiskLevel.NONE
|
||
assert result.risk_score < 20.0
|
||
assert "收入完整性检查通过" in result.description
|
||
assert result.risk_data["recharge_total"] == 45000.0
|
||
assert result.risk_data["declared_revenue"] == 45000.0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_revenue_match_without_contract(self, algorithm, mock_db_session, streamer_info,
|
||
mock_recharge_data, mock_declaration_data):
|
||
"""测试场景2：收入一致（无分成协议）"""
|
||
# 设置：充值总额45000元，无分成协议，申报45000元
|
||
|
||
# Mock数据库查询（无分成协议）
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(mock_recharge_data),
|
||
self._create_declaration_result(mock_declaration_data),
|
||
# 无分成协议查询，返回None
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言
|
||
assert result.risk_level == RiskLevel.NONE
|
||
assert result.risk_score < 20.0
|
||
assert "收入完整性检查通过" in result.description
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_slight_difference_low_risk(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景3：轻微差异（低风险）"""
|
||
# 充值45000元，申报44000元，差异1000元（差异率2.22%）
|
||
|
||
recharge_data = [
|
||
{"recharge_id": f"RC{i}", "user_name": "测试", "amount": 15000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
for i in range(1, 4)
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 44000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：差异在可接受范围内
|
||
assert result.risk_level == RiskLevel.NONE
|
||
assert 0 <= result.risk_score <= 20.0
|
||
|
||
|
||
class TestRevenueIntegrityUnderReporting(TestRevenueIntegrityAlgorithm):
|
||
"""少报收入场景测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_medium_under_reporting(self, algorithm, mock_db_session, streamer_info,
|
||
mock_contract_data):
|
||
"""测试场景4：中度少报收入（Medium风险）"""
|
||
# 充值100000元，分成70%，预期70000元，申报50000元，差异20000元（差异率28.57%）
|
||
|
||
recharge_data = [
|
||
{"recharge_id": f"RC{i}", "user_name": "测试", "amount": 20000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
for i in range(1, 6)
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 50000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
self._create_contract_result(mock_contract_data),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言
|
||
assert result.risk_level == RiskLevel.MEDIUM
|
||
assert 50.0 <= result.risk_score < 70.0
|
||
assert "隐瞒收入" in result.description or "少报" in result.description
|
||
assert result.risk_data["difference"] == 20000.0
|
||
assert 25.0 < result.risk_data["difference_rate"] < 30.0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_high_under_reporting(self, algorithm, mock_db_session, streamer_info,
|
||
mock_contract_data):
|
||
"""测试场景5：高度少报收入（High风险）"""
|
||
# 充值200000元，分成70%，预期140000元，申报80000元，差异60000元（差异率42.86%）
|
||
|
||
recharge_data = [
|
||
{"recharge_id": f"RC{i}", "user_name": "测试", "amount": 40000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
for i in range(1, 6)
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 80000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
self._create_contract_result(mock_contract_data),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言
|
||
assert result.risk_level == RiskLevel.HIGH
|
||
assert 70.0 <= result.risk_score < 85.0
|
||
assert "隐瞒收入" in result.description
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_critical_under_reporting(self, algorithm, mock_db_session, streamer_info,
|
||
mock_contract_data):
|
||
"""测试场景6：严重少报收入（Critical风险）"""
|
||
# 充值500000元，分成70%，预期350000元，申报100000元，差异250000元（差异率71.43%）
|
||
|
||
recharge_data = [
|
||
{"recharge_id": f"RC{i}", "user_name": "测试", "amount": 100000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
for i in range(1, 6)
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 100000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
self._create_contract_result(mock_contract_data),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言
|
||
assert result.risk_level == RiskLevel.CRITICAL
|
||
assert result.risk_score >= 85.0
|
||
assert "严重风险" in result.description
|
||
assert result.risk_data["difference"] >= 100000.0
|
||
|
||
|
||
class TestRevenueIntegrityOverReporting(TestRevenueIntegrityAlgorithm):
|
||
"""多报收入场景测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_over_reporting_low_risk(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景7：轻微多报收入（低风险）"""
|
||
# 充值40000元，申报41000元，多报1000元
|
||
|
||
recharge_data = [
|
||
{"recharge_id": f"RC{i}", "user_name": "测试", "amount": 20000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
for i in range(1, 3)
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 41000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：申报超过充值，但风险较低
|
||
assert result.risk_level == RiskLevel.LOW
|
||
assert "虚报" in result.description or "超过" in result.description
|
||
|
||
|
||
class TestRevenueIntegrityEdgeCases(TestRevenueIntegrityAlgorithm):
|
||
"""边界值和异常场景测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_recharge_data(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景8：无充值数据"""
|
||
# 无充值数据，有申报数据
|
||
|
||
recharge_data = [] # 空充值
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 10000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：无充值但有申报，风险很高
|
||
assert result.risk_level in [RiskLevel.HIGH, RiskLevel.CRITICAL]
|
||
assert "申报收入超过充值" in result.description
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_declaration_data(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景9：无申报数据"""
|
||
# 有充值数据，无申报数据
|
||
|
||
recharge_data = [
|
||
{"recharge_id": "RC001", "user_name": "测试", "amount": 10000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
]
|
||
|
||
declaration_data = [] # 空申报
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：有充值但无申报，严重风险
|
||
assert result.risk_level == RiskLevel.CRITICAL
|
||
assert "隐瞒收入" in result.description
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_zero_difference(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景10：完全一致（零差异）"""
|
||
# 充值与申报完全一致
|
||
|
||
recharge_data = [
|
||
{"recharge_id": "RC001", "user_name": "测试", "amount": 30000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 30000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：完全一致，无风险
|
||
assert result.risk_level == RiskLevel.NONE
|
||
assert result.risk_score == 0.0
|
||
assert "基本一致" in result.description
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_large_amount_boundary(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景11：大额边界值（刚好10万元差异）"""
|
||
# 充值200000元，申报100000元，差异100000元，差异率50%
|
||
|
||
recharge_data = [
|
||
{"recharge_id": "RC001", "user_name": "测试", "amount": 200000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 100000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：刚好达到Critical阈值
|
||
assert result.risk_level == RiskLevel.CRITICAL
|
||
assert result.risk_data["difference"] == 100000.0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_rate_boundary_50_percent(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景12：差异率边界值（刚好50%）"""
|
||
# 充值200000元，申报100000元，差异100000元，差异率50%
|
||
|
||
recharge_data = [
|
||
{"recharge_id": "RC001", "user_name": "测试", "amount": 200000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01", "sales_revenue": 100000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：50%差异率应达到Critical
|
||
assert result.risk_level == RiskLevel.CRITICAL
|
||
assert result.risk_data["difference_rate"] == 50.0
|
||
|
||
|
||
class TestRevenueIntegrityErrorHandling(TestRevenueIntegrityAlgorithm):
|
||
"""错误处理测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_missing_streamer_id(self, algorithm, mock_db_session):
|
||
"""测试场景13：缺少主播ID"""
|
||
context = self._create_context("", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：返回错误结果
|
||
assert result.risk_level == RiskLevel.UNKNOWN
|
||
assert "失败" in result.description
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_missing_period(self, algorithm, mock_db_session):
|
||
"""测试场景14：缺少期间参数"""
|
||
context = DetectionContext(
|
||
task_id="task_001",
|
||
rule_id="rule_001",
|
||
parameters={"streamer_id": "ZB_TEST_001"},
|
||
db_session=mock_db_session
|
||
)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：返回错误结果
|
||
assert result.risk_level == RiskLevel.UNKNOWN
|
||
assert "失败" in result.description
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_invalid_period_format(self, algorithm, mock_db_session):
|
||
"""测试场景15：无效的期间格式"""
|
||
context = self._create_context("ZB_TEST_001", "2024-13", mock_db_session) # 13月无效
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：返回错误结果
|
||
assert result.risk_level == RiskLevel.UNKNOWN
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_streamer_not_found(self, algorithm, mock_db_session):
|
||
"""测试场景16：主播不存在"""
|
||
# Mock主播查询返回None
|
||
mock_db_session.execute.side_effect = [
|
||
MagicMock(scalar_one_or_none=lambda: None), # 主播不存在
|
||
]
|
||
|
||
context = self._create_context("ZB_NONEXIST", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：返回错误结果
|
||
assert result.risk_level == RiskLevel.UNKNOWN
|
||
assert "找不到主播信息" in result.description
|
||
|
||
|
||
class TestRevenueIntegrityBusinessScenarios(TestRevenueIntegrityAlgorithm):
|
||
"""真实业务场景测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_monthly_reconciliation(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景17：月度对账场景"""
|
||
# 模拟真实月度对账：充值45万，申报30万，差异15万，需要核查
|
||
|
||
recharge_data = [
|
||
{
|
||
"recharge_id": f"RC{i:03d}",
|
||
"user_name": "测试主播",
|
||
"amount": 15000.0,
|
||
"time": datetime(2024, 1, 1) + timedelta(days=i),
|
||
"payment_method": "bank_transfer"
|
||
}
|
||
for i in range(1, 31) # 每天1笔，共30笔
|
||
]
|
||
|
||
declaration_data = [
|
||
{
|
||
"declaration_id": "TAX001",
|
||
"taxpayer_name": "测试主播",
|
||
"tax_period": "2024-01",
|
||
"sales_revenue": 300000.0,
|
||
"declaration_date": date(2024, 2, 15)
|
||
}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_TEST_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言
|
||
assert result.risk_level in [RiskLevel.HIGH, RiskLevel.CRITICAL]
|
||
assert result.risk_data["recharge_count"] == 30
|
||
assert result.risk_data["difference"] == 150000.0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_new_streamer_no_history(self, algorithm, mock_db_session, streamer_info):
|
||
"""测试场景18：新主播无历史数据"""
|
||
# 新主播，只有少量充值和申报
|
||
|
||
recharge_data = [
|
||
{"recharge_id": "RC001", "user_name": "新主播", "amount": 5000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "新主播",
|
||
"tax_period": "2024-01", "sales_revenue": 4800.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_NEW_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：新主播轻微差异也属正常
|
||
assert result.risk_level in [RiskLevel.LOW, RiskLevel.NONE]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_enterprise_streamer(self, algorithm, mock_db_session):
|
||
"""测试场景19：企业主播"""
|
||
# 企业主播，统一社会信用代码
|
||
|
||
streamer_info = {
|
||
"streamer_id": "ZB_ENT_001",
|
||
"streamer_name": "XX传媒有限公司",
|
||
"entity_type": "enterprise",
|
||
"tax_registration_no": None,
|
||
"unified_social_credit_code": "91110000000000000X",
|
||
"id_card_no": None,
|
||
}
|
||
|
||
recharge_data = [
|
||
{"recharge_id": "RC001", "user_name": "企业主播", "amount": 100000.0,
|
||
"time": datetime(2024, 1, 15), "payment_method": "bank_transfer"}
|
||
]
|
||
|
||
declaration_data = [
|
||
{"declaration_id": "TAX001", "taxpayer_name": "XX传媒有限公司",
|
||
"tax_period": "2024-01", "sales_revenue": 100000.0,
|
||
"declaration_date": date(2024, 2, 15)}
|
||
]
|
||
|
||
mock_db_session.execute.side_effect = [
|
||
self._create_streamer_result(streamer_info),
|
||
self._create_recharge_result(recharge_data),
|
||
self._create_declaration_result(declaration_data),
|
||
MagicMock(scalar_one_or_none=lambda: None),
|
||
]
|
||
|
||
# 执行检测
|
||
context = self._create_context("ZB_ENT_001", "2024-01", mock_db_session)
|
||
result = await algorithm.detect(context)
|
||
|
||
# 断言：企业主播数据一致
|
||
assert result.risk_level == RiskLevel.NONE
|
||
|
||
|
||
# ==================== 辅助方法 ====================
|
||
|
||
def _create_context(self, streamer_id: str, period: str, db_session: AsyncSession) -> DetectionContext:
|
||
"""创建检测上下文"""
|
||
return DetectionContext(
|
||
task_id="task_test_001",
|
||
rule_id="rule_test_001",
|
||
parameters={
|
||
"streamer_id": streamer_id,
|
||
"period": period,
|
||
},
|
||
db_session=db_session
|
||
)
|
||
|
||
def _create_streamer_result(self, streamer_info: dict) -> MagicMock:
|
||
"""创建主播查询结果"""
|
||
mock_streamer = MagicMock()
|
||
mock_streamer.streamer_id = streamer_info["streamer_id"]
|
||
mock_streamer.streamer_name = streamer_info["streamer_name"]
|
||
mock_streamer.entity_type = streamer_info["entity_type"]
|
||
mock_streamer.tax_registration_no = streamer_info["tax_registration_no"]
|
||
mock_streamer.unified_social_credit_code = streamer_info["unified_social_credit_code"]
|
||
mock_streamer.id_card_no = streamer_info["id_card_no"]
|
||
|
||
result = MagicMock()
|
||
result.scalar_one_or_none.return_value = mock_streamer
|
||
return result
|
||
|
||
def _create_recharge_result(self, recharge_data: list) -> MagicMock:
|
||
"""创建充值查询结果"""
|
||
mock_recharges = []
|
||
for data in recharge_data:
|
||
mock_recharge = MagicMock()
|
||
mock_recharge.recharge_id = data["recharge_id"]
|
||
mock_recharge.user_name = data["user_name"]
|
||
mock_recharge.actual_amount_cny = data["amount"]
|
||
mock_recharge.recharge_time = data["time"]
|
||
mock_recharge.payment_method = data["payment_method"]
|
||
mock_recharge.status = "success"
|
||
mock_recharges.append(mock_recharge)
|
||
|
||
# Mock汇总查询
|
||
summary = MagicMock()
|
||
summary.count = len(recharge_data)
|
||
summary.total = sum(r["amount"] for r in recharge_data)
|
||
|
||
result = MagicMock()
|
||
result.one.return_value = summary
|
||
result.scalars.return_value.all.return_value = mock_recharges
|
||
return result
|
||
|
||
def _create_declaration_result(self, declaration_data: list) -> MagicMock:
|
||
"""创建税务申报查询结果"""
|
||
mock_declarations = []
|
||
for data in declaration_data:
|
||
mock_decl = MagicMock()
|
||
mock_decl.vat_declaration_id = data["declaration_id"]
|
||
mock_decl.taxpayer_name = data["taxpayer_name"]
|
||
mock_decl.tax_period = data["tax_period"]
|
||
mock_decl.sales_revenue = data["sales_revenue"]
|
||
mock_decl.declaration_date = data["declaration_date"]
|
||
mock_declarations.append(mock_decl)
|
||
|
||
result = MagicMock()
|
||
result.scalars.return_value.all.return_value = mock_declarations
|
||
return result
|
||
|
||
def _create_contract_result(self, contract_data: dict) -> MagicMock:
|
||
"""创建分成协议查询结果"""
|
||
mock_contract = MagicMock()
|
||
mock_contract.streamer_ratio = contract_data["streamer_ratio"]
|
||
|
||
result = MagicMock()
|
||
result.scalar_one_or_none.return_value = mock_contract
|
||
return result
|
||
|
||
|
||
# ==================== 运行测试 ====================
|
||
|
||
if __name__ == "__main__":
|
||
# 运行测试
|
||
pytest.main([__file__, "-v", "--tb=short"])
|