
关于
使用 pytest 的 Temporal 工作流全面测试方法,提供针对特定测试场景的渐进式资源
name: temporal-python-testing description: "使用 pytest 对 Temporal 工作流进行全面测试的方法,针对特定测试场景的渐进式资源披露。" risk: unknown source: community date_added: "2026-02-27"
Temporal Python 测试策略
使用 pytest 对 Temporal 工作流进行全面测试的方法,针对特定测试场景的渐进式资源披露。
在以下情况不要使用此技能
- 任务与 Temporal Python 测试策略无关
- 需要此范围之外的其他领域或工具
说明
- 明确目标、约束条件和所需输入。
- 应用相关最佳实践并验证结果。
- 提供可操作的步骤和验证方法。
- 如需详细示例,请打开
resources/implementation-playbook.md。
在以下情况使用此技能
- 单元测试工作流 - 使用时间跳过的快速测试
- 集成测试 - 使用模拟活动的工作流
- 回放测试 - 根据生产历史验证确定性
- 本地开发 - 设置 Temporal 服务器和 pytest
- CI/CD 集成 - 自动化测试管道
- 覆盖率策略 - 达到 ≥80% 测试覆盖率
测试理念
推荐方法(来源:docs.temporal.io/develop/python/testing-suite):
- 以集成测试为主
- 使用 pytest 配合异步 fixtures
- 时间跳过实现快速反馈(数月工作流 → 数秒完成)
- 模拟活动以隔离工作流逻辑
- 使用回放测试验证确定性
三种测试类型:
- 单元测试:使用时间跳过的工作流,使用 ActivityEnvironment 的活动
- 集成测试:使用模拟活动的 Worker
- 端到端测试:完整 Temporal 服务器配合真实活动(谨慎使用)
可用资源
此技能通过渐进式披露提供详细指导。根据您的测试需求加载特定资源:
单元测试资源
文件:resources/unit-testing.md
何时加载:隔离测试单个工作流或活动
包含:
- 使用时间跳过的 WorkflowEnvironment
- 用于活动测试的 ActivityEnvironment
- 长时间运行工作流的快速执行
- 手动时间推进模式
- pytest fixtures 和模式
集成测试资源
文件:resources/integration-testing.md
何时加载:测试具有模拟外部依赖的工作流
包含:
- 活动模拟策略
- 错误注入模式
- 多活动工作流测试
- Signal 和 Query 测试
- 覆盖率策略
回放测试资源
文件:resources/replay-testing.md
何时加载:验证确定性或部署工作流变更
包含:
- 确定性验证
- 生产历史回放
- CI/CD 集成模式
- 版本兼容性测试
本地开发资源
文件:resources/local-setup.md
何时加载:设置开发环境
包含:
- Docker Compose 配置
- pytest 设置和配置
- 覆盖率工具集成
- 开发工作流
快速入门指南
基本工作流测试
import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
@pytest.fixture
async def workflow_env():
env = await WorkflowEnvironment.start_time_skipping()
yield env
await env.shutdown()
@pytest.mark.asyncio
async def test_workflow(workflow_env):
async with Worker(
workflow_env.client,
task_queue="test-queue",
workflows=[YourWorkflow],
activities=[your_activity],
):
result = await workflow_env.client.execute_workflow(
YourWorkflow.run,
args,
id="test-wf-id",
task_queue="test-queue",
)
assert result == expected
基本活动测试
from temporalio.testing import ActivityEnvironment
async def test_activity():
env = ActivityEnvironment()
result = await env.run(your_activity, "test-input")
assert result == expected_output
覆盖率目标
推荐覆盖率(来源:docs.temporal.io 最佳实践):
- 工作流:≥80% 逻辑覆盖率
- 活动:≥80% 逻辑覆盖率
- 集成:使用模拟活动的关键路径
- 回放:部署前所有工作流版本
关键测试原则
- 时间跳过 - 数月工作流在数秒内完成测试
- 模拟活动 - 将工作流逻辑与外部依赖隔离
- 回放测试 - 部署前验证确定性
- 高覆盖率 - 生产工作流 ≥80% 目标
- 快速反馈 - 单元测试在毫秒内运行
如何使用资源
按需加载特定资源:
- "展示单元测试模式" → 加载
resources/unit-testing.md - "如何模拟活动?" → 加载
resources/integration-testing.md - "设置本地 Temporal 服务器" → 加载
resources/local-setup.md - "验证确定性" → 加载
resources/replay-testing.md
兼容工具
Claude CodeCursor
标签
测试

