deep-risk/backend/app/tests/conftest.py
2025-12-14 20:08:27 +08:00

417 lines
14 KiB
Python

"""
测试配置文件
用于配置pytest测试环境和共享fixtures
"""
import pytest
import asyncio
from datetime import datetime, date, timedelta
from decimal import Decimal
from unittest.mock import AsyncMock, MagicMock
@pytest.fixture(scope="session")
def event_loop():
"""创建事件循环供异步测试使用"""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
def test_data():
"""通用测试数据"""
return {
"streamer_id": "ZB_TEST_001",
"period": "2024-01",
"task_id": "task_test_001",
"rule_id": "rule_test_001",
}
@pytest.fixture
def mock_streamer_data():
"""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",
"phone_number": "13800138000",
"bank_account_no": "6222021234567890123",
"bank_name": "中国工商银行",
}
@pytest.fixture
def mock_mcn_data():
"""Mock MCN机构数据"""
return {
"mcn_id": "MCN001",
"mcn_name": "测试MCN机构",
"unified_social_credit_code": "91110000000000000X",
"legal_person_name": "法人代表",
"bank_account_no": "6222029876543210987",
"bank_name": "中国建设银行",
"status": "active",
}
@pytest.fixture
def mock_recharge_data():
"""Mock充值数据"""
return [
{
"recharge_id": "RC001",
"user_name": "测试用户",
"user_id": "ZB_TEST_001",
"amount": 10000.0,
"time": datetime(2024, 1, 15, 10, 30, 0),
"payment_method": "bank_transfer",
"transaction_no": "TXN001",
"platform_order_no": "PO001",
"actual_amount_cny": 10000.0,
"total_coins": 100000.0,
"status": "success",
},
{
"recharge_id": "RC002",
"user_name": "测试用户",
"user_id": "ZB_TEST_001",
"amount": 20000.0,
"time": datetime(2024, 1, 20, 14, 20, 0),
"payment_method": "bank_transfer",
"transaction_no": "TXN002",
"platform_order_no": "PO002",
"actual_amount_cny": 20000.0,
"total_coins": 200000.0,
"status": "success",
},
{
"recharge_id": "RC003",
"user_name": "测试用户",
"user_id": "ZB_TEST_001",
"amount": 15000.0,
"time": datetime(2024, 1, 25, 9, 15, 0),
"payment_method": "bank_transfer",
"transaction_no": "TXN003",
"platform_order_no": "PO003",
"actual_amount_cny": 15000.0,
"total_coins": 150000.0,
"status": "success",
},
]
@pytest.fixture
def mock_declaration_data():
"""Mock税务申报数据"""
return [
{
"declaration_id": "TAX001",
"taxpayer_name": "测试主播",
"taxpayer_id": "110101199001011234",
"tax_period": "2024-01",
"declaration_date": date(2024, 2, 15),
"tax_authority_code": "110101",
"tax_authority_name": "朝阳区税务局",
"taxpayer_type": "small_scale",
"tax_rate": 0.03,
"sales_revenue": 45000.0,
"sales_revenue_taxable": 43689.32,
"output_tax": 1310.68,
"tax_payable": 1310.68,
"tax_to_pay": 1310.68,
"declaration_status": "submitted",
}
]
@pytest.fixture
def mock_contract_data():
"""Mock分成协议数据"""
return {
"contract_id": "CT001",
"contract_no": "HT202401001",
"contract_type": "tip_sharing",
"streamer_id": "ZB_TEST_001",
"streamer_name": "测试主播",
"streamer_entity_type": "individual",
"platform_party": "XX科技有限公司",
"platform_credit_code": "91110000000000000X",
"revenue_type": "tip",
"streamer_ratio": 70.0,
"platform_ratio": 30.0,
"settlement_cycle": "monthly",
"contract_start_date": date(2024, 1, 1),
"contract_end_date": date(2024, 12, 31),
"contract_status": "active",
}
@pytest.fixture
def mock_bank_transaction_data():
"""Mock银行流水数据"""
return [
{
"transaction_id": "TXN001",
"account_no": "6222021234567890123",
"account_name": "测试主播",
"bank_name": "中国工商银行",
"transaction_date": date(2024, 1, 15),
"transaction_time": datetime(2024, 1, 15, 10, 30, 0),
"transaction_type": "转入",
"transaction_amount": 10000.0,
"balance": 50000.0,
"counterparty_account_no": "6222029876543210987",
"counterparty_account_name": "XX公司",
"counterparty_bank_name": "中国建设银行",
"voucher_no": "VZ001",
"transaction_purpose": "收入",
"is_cross_border": False,
"currency": "CNY",
"amount_cny": 10000.0,
"exchange_rate": 1.0,
"is_large_amount": False,
"is_suspicious": False,
}
]
@pytest.fixture
def mock_invoice_data():
"""Mock发票数据"""
return [
{
"invoice_id": "INV001",
"invoice_code": "011001900111",
"invoice_no": "12345678",
"invoice_type": "vat_general",
"direction": "issued",
"invoice_date": date(2024, 1, 20),
"purchaser_name": "YY贸易公司",
"purchaser_tax_no": "91110000000000001Y",
"seller_name": "测试主播",
"seller_tax_no": "110101199001011234",
"total_amount": 10000.0,
"total_tax": 600.0,
"total_amount_with_tax": 10600.0,
"amount_in_words": "壹万零陆佰元整",
"invoice_status": "normal",
"is_verified": True,
"is_red_invoice": False,
}
]
@pytest.fixture
def mock_expense_data():
"""Mock费用凭证数据"""
return [
{
"expense_id": "EXP001",
"voucher_no": "PZ202401001",
"expense_type": "advertising_fee",
"expense_category": "sales_expense",
"payer_name": "测试主播",
"payer_account_no": "6222021234567890123",
"payee_name": "XX广告公司",
"payee_account_no": "6222029876543210987",
"payee_bank_name": "中国建设银行",
"expense_date": date(2024, 1, 18),
"expense_amount": 5000.0,
"tax_amount": 300.0,
"tax_rate": 0.06,
"payment_method": "bank_transfer",
"payment_status": "paid",
"accounting_status": "posted",
"fiscal_year": 2024,
"fiscal_period": 1,
"is_large_amount": False,
"is_cross_border": False,
}
]
@pytest.fixture
def mock_order_data():
"""Mock订单数据"""
return [
{
"order_id": "ORD001",
"platform_order_no": "DD202401001234",
"ecommerce_platform": "douyin",
"streamer_id": "ZB_TEST_001",
"streamer_name": "测试主播",
"product_id": "SP001",
"product_name": "美妆护肤套装",
"quantity": 2,
"original_price": 299.0,
"sale_price": 199.0,
"total_amount": 398.0,
"actual_payment": 378.0,
"commission_ratio": 0.2,
"commission_amount": 75.6,
"streamer_commission": 52.92,
"buyer_id": "U123456",
"order_time": datetime(2024, 1, 15, 10, 30, 0),
"settlement_time": datetime(2024, 1, 20, 10, 0, 0),
"order_status": "completed",
"is_commission_settled": True,
}
]
@pytest.fixture
def mock_settlement_data():
"""Mock结算单数据"""
return [
{
"settlement_id": "SET001",
"settlement_no": "JS202401001",
"streamer_id": "ZB_TEST_001",
"streamer_name": "测试主播",
"streamer_entity_type": "individual",
"settlement_period": "2024-01",
"settlement_start_date": date(2024, 1, 1),
"settlement_end_date": date(2024, 1, 31),
"order_count": 100,
"total_sales": 50000.0,
"total_commission": 10000.0,
"platform_service_fee": 500.0,
"actual_settlement_amount": 9500.0,
"tax_withholding": 950.0,
"payment_method": "bank_transfer",
"payment_account_no": "6222021234567890123",
"payment_account_name": "测试主播",
"payment_time": datetime(2024, 2, 1, 10, 0, 0),
"payment_status": "paid",
"settlement_status": "settled",
}
]
# ==================== 风险等级测试数据 ====================
@pytest.fixture
def risk_level_test_cases():
"""风险等级测试用例数据"""
return {
"none": {
"recharge_total": 100000.0,
"declared_revenue": 100000.0,
"expected_risk_level": RiskLevel.NONE,
"description": "收入完全一致",
},
"low_under_reporting": {
"recharge_total": 100000.0,
"declared_revenue": 95000.0,
"expected_risk_level": RiskLevel.LOW,
"description": "轻微少报",
},
"medium_under_reporting": {
"recharge_total": 100000.0,
"declared_revenue": 85000.0,
"expected_risk_level": RiskLevel.MEDIUM,
"description": "中度少报",
},
"high_under_reporting": {
"recharge_total": 100000.0,
"declared_revenue": 65000.0,
"expected_risk_level": RiskLevel.HIGH,
"description": "高度少报",
},
"critical_under_reporting": {
"recharge_total": 100000.0,
"declared_revenue": 30000.0,
"expected_risk_level": RiskLevel.CRITICAL,
"description": "严重少报",
},
}
# ==================== 参数化测试数据 ====================
@pytest.fixture(params=[
# 参数化测试:不同分成比例
{"streamer_ratio": 50.0, "recharge_total": 100000.0, "expected_revenue": 50000.0},
{"streamer_ratio": 60.0, "recharge_total": 100000.0, "expected_revenue": 60000.0},
{"streamer_ratio": 70.0, "recharge_total": 100000.0, "expected_revenue": 70000.0},
{"streamer_ratio": 80.0, "recharge_total": 100000.0, "expected_revenue": 80000.0},
{"streamer_ratio": 90.0, "recharge_total": 100000.0, "expected_revenue": 90000.0},
])
def contract_ratio_param(request):
"""参数化测试:不同分成比例"""
return request.param
@pytest.fixture(params=[
# 参数化测试:不同金额规模
{"recharge_total": 1000.0, "threshold": 5000.0},
{"recharge_total": 10000.0, "threshold": 50000.0},
{"recharge_total": 100000.0, "threshold": 500000.0},
{"recharge_total": 1000000.0, "threshold": 5000000.0},
])
def amount_scale_param(request):
"""参数化测试:不同金额规模"""
return request.param
# ==================== 工具方法 ====================
def create_mock_streamer(streamer_data: dict) -> MagicMock:
"""创建主播Mock对象"""
mock = MagicMock()
mock.streamer_id = streamer_data["streamer_id"]
mock.streamer_name = streamer_data["streamer_name"]
mock.entity_type = streamer_data["entity_type"]
mock.tax_registration_no = streamer_data["tax_registration_no"]
mock.unified_social_credit_code = streamer_data["unified_social_credit_code"]
mock.id_card_no = streamer_data["id_card_no"]
mock.phone_number = streamer_data["phone_number"]
mock.bank_account_no = streamer_data["bank_account_no"]
mock.bank_name = streamer_data["bank_name"]
return mock
def create_mock_recharge(recharge_data: dict) -> MagicMock:
"""创建充值Mock对象"""
mock = MagicMock()
mock.recharge_id = recharge_data["recharge_id"]
mock.user_name = recharge_data["user_name"]
mock.user_id = recharge_data["user_id"]
mock.actual_amount_cny = recharge_data["amount"]
mock.recharge_time = recharge_data["time"]
mock.payment_method = recharge_data["payment_method"]
mock.transaction_no = recharge_data["transaction_no"]
mock.platform_order_no = recharge_data["platform_order_no"]
mock.total_coins = recharge_data["total_coins"]
mock.status = recharge_data["status"]
return mock
def create_mock_declaration(decl_data: dict) -> MagicMock:
"""创建申报Mock对象"""
mock = MagicMock()
mock.vat_declaration_id = decl_data["declaration_id"]
mock.taxpayer_name = decl_data["taxpayer_name"]
mock.taxpayer_id = decl_data["taxpayer_id"]
mock.tax_period = decl_data["tax_period"]
mock.sales_revenue = decl_data["sales_revenue"]
mock.declaration_date = decl_data["declaration_date"]
mock.tax_rate = decl_data["tax_rate"]
mock.output_tax = decl_data["output_tax"]
mock.tax_payable = decl_data["tax_payable"]
mock.declaration_status = decl_data["declaration_status"]
return mock
def create_mock_contract(contract_data: dict) -> MagicMock:
"""创建协议Mock对象"""
mock = MagicMock()
mock.streamer_ratio = contract_data["streamer_ratio"]
mock.platform_ratio = contract_data["platform_ratio"]
return mock