ZQYY.FQ-ETL/tests

tests/ — 测试套件

目录结构

tests/
├── unit/               # 单元测试（FakeDB/FakeAPI，无需真实数据库）
│   ├── task_test_utils.py   # 测试工具：FakeDBOperations、FakeAPIClient、OfflineAPIClient、TaskSpec
│   ├── test_ods_tasks.py    # ODS 任务在线/离线模式测试
│   ├── test_cli_args.py     # CLI 参数解析测试
│   ├── test_config.py       # 配置管理测试
│   ├── test_e2e_flow.py     # 端到端流程测试（CLI → PipelineRunner → TaskExecutor）
│   ├── test_task_registry.py        # 任务注册表测试
│   ├── test_*_properties.py         # 属性测试（hypothesis）
│   └── test_audit_*.py              # 仓库审计相关测试
└── integration/        # 集成测试（需要真实数据库）
    ├── test_database.py     # 数据库连接与操作测试
    └── test_index_tasks.py  # 指数任务集成测试

运行测试

# 安装测试依赖
pip install pytest hypothesis

# 全部单元测试
pytest tests/unit

# 指定测试文件
pytest tests/unit/test_ods_tasks.py

# 按关键字过滤
pytest tests/unit -k "online"

# 集成测试（需要设置 TEST_DB_DSN）
TEST_DB_DSN="postgresql://user:pass@host:5432/db" pytest tests/integration

# 查看详细输出
pytest tests/unit -v --tb=short

测试工具（task_test_utils.py）

单元测试通过 tests/unit/task_test_utils.py 提供的桩对象避免依赖真实数据库和 API：

• FakeDBOperations — 拦截并记录 upsert/execute/commit/rollback，不触碰真实数据库
• FakeAPIClient — 在线模式桩，直接返回预置的内存数据
• OfflineAPIClient — 离线模式桩，从归档 JSON 文件回放数据
• TaskSpec — 描述任务测试元数据（任务代码、端点、数据路径、样例记录）
• create_test_config() — 构建测试用 AppConfig
• dump_offline_payload() — 将样例数据写入归档目录供离线测试使用

编写新测试

• 单元测试放在 tests/unit/，文件名 test_*.py
• 使用 FakeDBOperations 和 FakeAPIClient 避免外部依赖
• 属性测试使用 hypothesis，文件名以 _properties.py 结尾
• 集成测试放在 tests/integration/，通过 TEST_DB_DSN 环境变量控制是否执行