🇨🇳 简体中文
🇺🇸 English
🇯🇵 日本語

Fire 项目测试套件

📁 测试文件结构

tests/
├── conftest.py                         # Pytest配置和fixtures
├── pytest.ini                        # Pytest设置和自定义标记
├── pytest_parallel.ini               # 并行测试专用配置
├── requirements-test.txt              # 测试依赖
├── unit/                             # 单元测试
│   ├── test_fee_calculator.py             # 交易费用计算器测试
│   ├── test_data_source_architecture.py   # 统一数据源架构测试
│   ├── test_environment_fallback.py       # 环境回退机制测试
│   ├── test_optimized_data_source.py      # 数据源优化测试
│   └── test_backtest_future_function.py   # 回测未来函数防护测试
├── trading/                          # 交易模块测试
│   ├── test_simulation_engine.py          # 模拟交易引擎测试
│   └── test_risk_engine.py                # 风险引擎测试
├── integration/                      # 集成测试
│   ├── test_trading_flow.py               # 完整交易流程测试
│   └── test_backtest_future_function_integration.py  # 回测集成测试
└── e2e/                             # 端到端测试
    ├── test_data_source_api.py            # 数据源API测试
    └── test_trade_api.py                  # 交易API测试

🧪 测试文件详解

单元测试 (Unit Tests)

test_fee_calculator.py

测试内容: 交易费用计算器 代码关联:

test_data_source_architecture.py

测试内容: 统一数据源架构适配器 代码关联:

test_environment_fallback.py

测试内容: 环境回退机制 代码关联:

test_optimized_data_source.py

测试内容: 数据源优化 代码关联:

test_backtest_future_function.py

测试内容: 回测未来函数防护 代码关联:

交易模块测试 (Trading Tests)

test_simulation_engine.py

测试内容: 模拟交易引擎 代码关联:

test_risk_engine.py

测试内容: 风险引擎 代码关联:

集成测试 (Integration Tests)

test_trading_flow.py

测试内容: 完整交易流程 代码关联:

test_backtest_future_function_integration.py

测试内容: 回测未来函数集成测试 代码关联:

端到端测试 (E2E Tests)

test_data_source_api.py

测试内容: 数据源API端点 代码关联:

test_trade_api.py

测试内容: 交易测试模块的数据获取API 代码关联:

测试范围: 仅测试数据获取相关功能，不测试交易提交功能 测试的API:

不测试的API:

🚀 运行测试

基本命令

# 进入backend目录
cd backend
source venv/bin/activate

# 运行所有测试
pytest tests/ -v

# 运行特定类型测试
pytest tests/unit/ -v    # 单元测试
pytest tests/trading/ -v # 交易模块测试
pytest tests/integration/ -v # 集成测试
pytest tests/e2e/ -v     # 端到端测试

# 运行特定文件
pytest tests/unit/test_fee_calculator.py -v
pytest tests/e2e/test_data_source_api.py -v

并行测试

快速开始

# 运行单元测试（推荐）
python tests/run_tests.py --unit

# 运行所有测试
python tests/run_tests.py --all

# 查看执行计划
python tests/test_dependencies.py

使用pytest直接执行

# 并行执行单元测试
pytest tests/unit/ -n auto --dist=loadscope

# 并行执行指定测试
pytest tests/unit/test_fee_calculator.py tests/unit/test_environment_fallback.py -n 2

# 串行执行集成测试
pytest tests/integration/ -v

高级并行执行

# 使用高级并行执行器
python tests/run_parallel_tests.py --plan  # 查看执行计划
python tests/run_parallel_tests.py --tests test_fee_calculator test_environment_fallback
python tests/run_parallel_tests.py  # 运行所有测试

测试配置

测试用户: test_for_dev / test_only_for_dev 测试环境: ENVIRONMENT=test 测试数据库: Redis DB 15 JWT Token: 开发环境永不过期

高级选项

# 带覆盖率报告
pytest tests/ --cov=. --cov-report=html

# 详细输出
pytest tests/ -v -s

# 只运行失败的测试
pytest tests/ --lf

# 在第一个失败时停止
pytest tests/ -x

# 显示最慢的10个测试
pytest tests/ --durations=10

# 并行执行（自动检测CPU核心数）
pytest tests/unit/ -n auto

# 并行执行（指定进程数）
pytest tests/unit/ -n 4

📊 测试覆盖范围

API端点测试

架构组件测试

不测试的模块

🔧 并行测试系统

测试分类

1. 单元测试 (Unit Tests)

2. 集成测试 (Integration Tests)

3. E2E测试 (End-to-End Tests)

依赖管理

测试依赖关系

第1轮 (可并行):
├── test_fee_calculator
├── test_data_source_architecture
├── test_environment_fallback
├── test_optimized_data_source
├── test_backtest_future_function
├── test_simulation_engine
└── test_risk_engine

第2轮 (部分并行):
├── test_trading_flow (依赖: test_simulation_engine)
├── test_backtest_future_function_integration (依赖: test_backtest_future_function)
└── test_data_source_api (依赖: test_data_source_architecture)

第3轮 (串行):
└── test_trade_api (依赖: test_trading_flow)

资源隔离

性能优化

预期性能提升

实际测试结果

# 串行执行
pytest tests/unit/ -v  # 约15-20秒

# 并行执行
pytest tests/unit/ -n auto  # 约5-8秒

配置选项

pytest.ini 配置

[pytest]
addopts = 
    -v
    --tb=short
    --strict-markers
    --disable-warnings
markers =
    unit: Unit tests
    integration: Integration tests
    e2e: End-to-end tests
    redis_dependent: Tests that depend on Redis
    session_dependent: Tests that depend on trading session
    database_dependent: Tests that depend on database
    sequential: Tests that must run sequentially
    slow: Slow running tests

并行参数说明

🛠️ 故障排除

常见问题

  1. 模块导入错误 
    cd backend
source venv/bin/activate
  2. 测试依赖缺失 
    pip install -r tests/requirements-test.txt
  3. 数据源架构导入错误 
    pip install -e .
  4. 测试冲突
    • 确保每个测试使用独立的资源（session_id, redis数据库等）
    • 检查是否有共享状态
  5. 依赖问题
    • 使用 python tests/test_dependencies.py 查看依赖关系
    • 确保依赖的测试先执行
  6. 性能问题
    • 避免在并行测试中使用 time.sleep()
    • 使用mock替代真实的外部服务

调试命令

# 查看详细的并行执行信息
pytest tests/unit/ -n 2 -v -s

# 查看测试收集信息
pytest tests/unit/ --collect-only

# 运行单个测试进行调试
pytest tests/unit/test_fee_calculator.py::TestFeeCalculator::test_us_buy_basic -v -s

# 查看测试执行计划
python tests/test_dependencies.py

🔧 测试配置

测试标记

测试环境

📝 测试模板

import pytest
from unittest.mock import Mock, patch
from core.data_source.adapters.asset_adapter import AssetAdapter

@pytest.mark.unit
@pytest.mark.redis_dependent
class TestDataSourceArchitecture:
    """数据源架构测试"""
    
    def setup_method(self):
        """每个测试前的设置"""
        # 使用唯一的session_id避免冲突
        self.session_id = f"test_session_{uuid.uuid4().hex[:8]}"
        self.redis_db = 15  # 测试专用
    
    def test_adapter_initialization(self):
        """测试适配器初始化"""
        adapter = AssetAdapter("test_user_123")
        assert adapter.user_id == "test_user_123"
    
    @patch('core.data_source.adapters.asset_adapter.AssetDataSourceAdapter')
    def test_adapter_functionality(self, mock_adapter):
        """测试适配器功能"""
        mock_adapter.return_value.get_positions.return_value = []
        adapter = AssetAdapter("test_user_123")
        result = adapter.get_positions()
        assert result is not None

🎯 最佳实践

1. 编写并行安全的测试

import uuid

class TestSimulationEngine:
    def setup_method(self):
        # 使用唯一的session_id避免冲突
        self.session_id = f"test_session_{uuid.uuid4().hex[:8]}"
        self.engine = SimulationEngine(self.user_id, self.session_id, self.initial_capital)

2. 资源隔离

@pytest.fixture(autouse=True)
def setup_test_environment(mock_redis):
    """自动设置测试环境"""
    # 在每个测试前清理Redis
    mock_redis.flushdb()
    yield
    # 测试后清理
    mock_redis.flushdb()

3. 标记测试

@pytest.mark.unit
@pytest.mark.redis_dependent
class TestDataSourceArchitecture:
    """数据源架构测试"""
    pass