2025
请求返回一部分内容后中断,但服务器本身没有主动断开连接
git config pull.rebase false
20250514
UUID vs 雪花ID Base62 编码:唯一标识符在系统设计中的全面对比与选型指南
OpenAPI 到 FastAPI 服务端代码生成项目
ollama Segmentation fault
在 Temporal Python SDK 中系统化管理 RetryPolicy
[cursor-plan] budget_dayparting 模块测试方案
本文档使用 MrDoc 发布
-
+
首页
[cursor-plan] budget_dayparting 模块测试方案
# budget_dayparting 模块测试方案 ## 一、测试基础设施搭建 ### 1.1 测试依赖安装 在 `pyproject.toml` 中添加测试依赖: - `pytest>=8.0.0` - 测试框架 - `pytest-asyncio>=0.23.0` - 异步测试支持 - `pytest-mock>=3.12.0` - Mock 工具 - `pytest-cov>=4.1.0` - 代码覆盖率 - `freezegun>=1.4.0` - 时间控制 - `faker>=22.0.0` - 测试数据生成 ### 1.2 测试配置文件 创建 `tests/conftest.py` 作为全局 pytest 配置: - 配置 pytest-asyncio 模式 - 设置测试环境变量 - 创建全局 fixtures(测试数据库连接、Mock Amazon API 客户端、Mock Temporal context) ### 1.3 测试数据库初始化 创建 `tests/fixtures/db_fixtures.py`: - `test_db_connection` fixture:创建测试数据库连接池 - `db_transaction` fixture:每个测试使用独立事务,测试后自动回滚 - `sample_dayparting_data` fixture:生成标准测试数据(action 配置、campaign、portfolio) 创建 `tests/fixtures/db_schema.sql`: - 定义测试所需的数据库表结构(从实际 schema 简化) - 包含核心表:`hanna_action_budget_dayparting`, `hanna_action_budget_dayparting_record`, `hanna_action_budget_dayparting_day_detail`, `hanna_action_budget_dayparting_hour_detail`, `amz_ads_data.sl_amz_campaigns_dim_basic` ### 1.4 Amazon API Mock 创建 `tests/mocks/amazon_api_mock.py`: - `MockAmazonAdvertisingClient` 类:模拟 AmazonAdvertisingClient 的所有接口 - 模拟 SP/SB/SD campaign update 接口 - 模拟 portfolio update 接口 - 模拟 budget rule 关联/解除接口 - 记录所有 API 调用以便验证 ### 1.5 Temporal 测试环境 创建 `tests/fixtures/temporal_fixtures.py`: - 使用 `temporalio.testing.WorkflowEnvironment` 创建测试环境 - 配置测试用的 Activity 和 Workflow - `mock_activity_context` fixture:模拟 activity.logger 等上下文 ## 二、单元测试(tests/unit/) ### 2.1 数据库服务测试(test_db_services.py) **DaypartingService 测试:** - `test_get_dayparting_data_success`:测试正常获取分时预算配置 - `test_get_dayparting_data_not_found`:测试配置不存在的情况 - `test_update_dayparting_run_status_completed`:测试更新为 COMPLETED 状态 - `test_update_dayparting_run_status_failed`:测试更新为 FAILED 状态 - `test_update_dayparting_run_status_pending`:测试更新为 PENDING 状态 **DaypartingRecordService 测试:** - `test_get_pending_record_target_ids`:测试获取待处理记录 - `test_create_or_update_record_create`:测试创建新记录 - `test_create_or_update_record_update`:测试更新现有记录 - `test_mark_records_completed_single`:测试标记单个记录完成 - `test_mark_records_completed_batch`:测试批量标记完成 - `test_mark_records_completed_empty`:测试空列表情况 **DaypartingHourDetailService 测试:** - `test_check_action_is_executed_true`:测试已执行检查 - `test_check_action_is_executed_false`:测试未执行检查 - `test_get_executed_target_ids`:测试获取已执行目标 - `test_get_or_create_hour_detail_record_create`:测试创建小时记录 - `test_get_or_create_hour_detail_record_exists`:测试获取已存在记录 - `test_get_latest_budget_record`:测试获取最新预算记录 - `test_update_hour_detail_status`:测试更新小时记录状态(级联更新) **DaypartingTakeoverService 测试:** - `test_get_target_migration_records`:测试获取迁移记录 - `test_get_target_migration_records_empty`:测试无迁移记录情况 **AMZCampaignsDimBasicService 测试:** - `test_get_campaign_basic_info`:测试获取广告活动基础信息 - `test_get_campaign_basic_info_multiple`:测试批量获取 ### 2.2 规则活动测试(test_budget_dayparting_rule_activity.py) **配置验证测试:** - `test_validate_rule_config_success`:测试有效配置 - `test_validate_rule_config_missing_weekly_metrics`:测试缺少周度指标 - `test_validate_rule_config_invalid_weekly_metrics_length`:测试周度指标长度错误 - `test_validate_rule_config_missing_start_time`:测试缺少开始时间 - `test_validate_rule_config_invalid_dayparting_type`:测试无效分时类型 - `test_validate_rule_config_invalid_budget_strategy`:测试无效预算策略 **规则有效性测试:** - `test_is_dayparting_rule_valid_active`:测试有效规则 - `test_is_dayparting_rule_valid_deleted`:测试已删除规则 - `test_is_dayparting_rule_valid_inactive`:测试未激活规则 - `test_is_dayparting_rule_valid_expired`:测试已过期规则 - `test_is_dayparting_rule_valid_not_started`:测试未开始规则 **预算调整逻辑测试:** - `test_execute_dayparting_adjustments_fixed_budget`:测试固定预算调整 - `test_execute_dayparting_adjustments_percentage_budget`:测试百分比预算调整 - `test_execute_dayparting_adjustments_paused`:测试暂停状态调整 - `test_execute_dayparting_adjustments_rollback`:测试预算回退 - `test_calculate_baseline_budget_no_history`:测试无历史记录时的基准预算 - `test_calculate_baseline_budget_with_history`:测试有历史记录时的基准预算 - `test_calculate_baseline_budget_inconsistent`:测试数据不一致情况 **Campaign 相关测试:** - `test_fetch_campaign_details_sp`:测试获取 SP campaign 详情 - `test_fetch_campaign_details_sb`:测试获取 SB campaign 详情 - `test_fetch_campaign_details_sd`:测试获取 SD campaign 详情 - `test_fetch_campaign_details_with_budget_rules`:测试包含预算规则的 campaign - `test_calc_new_budget_within_limits`:测试预算限制计算(正常范围) - `test_calc_new_budget_below_min`:测试预算低于最小值 - `test_calc_new_budget_above_max`:测试预算高于最大值 **Portfolio 相关测试:** - `test_fetch_portfolio_details`:测试获取投资组合详情 - `test_create_target_data_portfolio`:测试创建投资组合数据 - `test_create_target_data_campaign`:测试创建广告活动数据 **接管逻辑测试:** - `test_get_target_migration_lists_no_migration`:测试无迁移情况 - `test_get_target_migration_lists_migrated_out`:测试迁出场景 - `test_get_target_migration_lists_migrated_in`:测试迁入场景 - `test_get_target_migration_lists_complex_chain`:测试复杂迁移链 **执行流程测试:** - `test_run_action_already_executed`:测试本小时已执行 - `test_run_action_no_pending_targets`:测试无待处理目标 - `test_run_action_valid_rule`:测试有效规则执行 - `test_run_action_invalid_rule`:测试无效规则回滚 ### 2.3 推送活动测试(test_push_budget_dayparting_activity.py) **基础流程测试:** - `test_push_activity_already_completed`:测试已完成任务跳过 - `test_push_activity_missing_target_ids`:测试缺少目标 ID **Campaign 推送测试:** - `test_update_campaign_budget_sp_success`:测试 SP campaign 预算更新成功 - `test_update_campaign_budget_sb_success`:测试 SB campaign 预算更新成功 - `test_update_campaign_budget_sd_success`:测试 SD campaign 预算更新成功 - `test_update_campaign_budget_with_budget_rules`:测试包含预算规则的更新 - `test_update_campaign_budget_remove_budget_rules`:测试解除预算规则 - `test_update_campaign_budget_api_failure`:测试 API 调用失败 **Portfolio 推送测试:** - `test_update_portfolio_budget_success`:测试投资组合预算更新成功 - `test_update_portfolio_budget_no_cap`:测试 NO_CAP 策略 - `test_update_portfolio_budget_api_failure`:测试 API 失败 **状态更新测试:** - `test_push_activity_success_updates_status`:测试成功后状态更新 - `test_push_activity_failure_updates_status`:测试失败后状态更新 ### 2.4 记录更新活动测试(test_update_record_activity.py) - `test_update_record_activity_with_next_run_time`:测试设置下次运行时间 - `test_update_record_activity_without_next_run_time`:测试无下次运行时间 ## 三、集成测试(tests/integration/) ### 3.1 端到端流程测试(test_dayparting_e2e.py) **完整执行流程:** - `test_campaign_dayparting_full_cycle`:测试 campaign 分时预算完整周期 - 规则活动创建记录 - 推送活动执行预算调整 - 验证数据库状态正确 - 验证 API 调用正确 - `test_portfolio_dayparting_full_cycle`:测试 portfolio 分时预算完整周期 - `test_dayparting_rollback_cycle`:测试预算回退完整流程 - 第一小时应用预算 - 第二小时回退预算 - 验证数据一致性 - `test_dayparting_with_takeover`:测试接管场景 - Action A 执行 - Action B 接管部分 targets - Action A 正确处理迁出 - Action B 正确处理迁入 **多小时连续执行:** - `test_dayparting_multiple_hours`:测试连续多个小时执行 - 模拟 24 小时周期 - 验证每小时独立执行 - 验证基准预算计算正确 **边界场景:** - `test_dayparting_across_days`:测试跨天执行 - `test_dayparting_timezone_handling`:测试时区处理 - `test_dayparting_during_rule_expiration`:测试规则到期时的处理 ### 3.2 数据库事务测试(test_db_transactions.py) - `test_transaction_rollback_on_error`:测试错误时事务回滚 - `test_concurrent_updates_handling`:测试并发更新处理 - `test_cascade_status_updates`:测试级联状态更新 ## 四、性能测试(tests/performance/) ### 4.1 批量处理测试(test_batch_processing.py) - `test_batch_campaign_processing`:测试批量处理 100+ campaigns - `test_batch_portfolio_processing`:测试批量处理 50+ portfolios - `test_database_query_performance`:测试数据库查询性能 ## 五、测试数据生成器(tests/factories/) ### 5.1 数据工厂(test_data_factories.py) - `ActionFactory`:生成 action 配置数据 - `CampaignFactory`:生成 campaign 测试数据 - `PortfolioFactory`:生成 portfolio 测试数据 - `WeeklyMetricsFactory`:生成周度指标配置 ## 六、测试运行配置 ### 6.1 pytest.ini 配置文件包含: - 测试目录:`tests/` - 异步模式:`asyncio_mode = auto` - 覆盖率配置:最低 80% 覆盖率 - 日志配置 - 标记定义:`unit`, `integration`, `e2e`, `slow` ### 6.2 运行命令 ```bash # 运行所有测试 uv run pytest # 运行单元测试 uv run pytest tests/unit -v # 运行集成测试 uv run pytest tests/integration -v # 生成覆盖率报告 uv run pytest --cov=src/activities/budget_dayparting --cov-report=html # 运行特定测试 uv run pytest tests/unit/test_db_services.py::test_get_dayparting_data_success -v ``` ## 七、核心测试策略 ### 7.1 Mock 策略 - **Amazon API**:全部 Mock,记录调用以验证 - **数据库**:使用真实测试数据库(PostgreSQL),每个测试独立事务 - **Temporal Context**:Mock activity.logger 和相关上下文 - **时间控制**:使用 freezegun 固定时间 ### 7.2 测试数据隔离 - 每个测试使用独立的数据库事务 - 测试结束自动回滚 - 使用唯一的 UUID 避免数据冲突 ### 7.3 断言验证 - 数据库状态验证(记录是否正确创建/更新) - API 调用验证(Mock 记录的调用次数、参数) - 返回值验证(结果结构、状态码) - 异常验证(错误场景是否正确抛出异常) ## 八、文件结构 ``` tests/ ├── conftest.py # 全局配置 ├── pytest.ini # pytest 配置 ├── __init__.py ├── fixtures/ │ ├── __init__.py │ ├── db_fixtures.py # 数据库 fixtures │ ├── temporal_fixtures.py # Temporal fixtures │ └── db_schema.sql # 测试数据库 schema ├── mocks/ │ ├── __init__.py │ └── amazon_api_mock.py # Amazon API Mock ├── factories/ │ ├── __init__.py │ └── test_data_factories.py # 测试数据生成器 ├── unit/ │ ├── __init__.py │ ├── test_db_services.py # 数据库服务单元测试 │ ├── test_budget_dayparting_rule_activity.py # 规则活动单元测试 │ ├── test_push_budget_dayparting_activity.py # 推送活动单元测试 │ └── test_update_record_activity.py # 记录更新单元测试 ├── integration/ │ ├── __init__.py │ ├── test_dayparting_e2e.py # 端到端集成测试 │ └── test_db_transactions.py # 数据库事务测试 └── performance/ ├── __init__.py └── test_batch_processing.py # 批量处理性能测试 ```
幻翼
2025年11月5日 10:12
转发文档
收藏文档
上一篇
下一篇
手机扫码
复制链接
手机扫一扫转发分享
复制链接
Markdown文件
PDF文档(打印)
分享
链接
类型
密码
更新密码