关于 temporal-python-testing
面向使用 Python SDK 开发 Temporal 工作流的开发者,提供基于 pytest 的分层测试策略,涵盖时间跳跃、模拟 Activity 和重放验证。
主要功能
- 01工作流与 Activity 单元测试
- 02时间跳跃测试长流程
- 03模拟 Activity 的集成测试
- 04重放测试与确定性验证
- 05本地开发和 CI/CD 配置指导
适用场景
- 01隔离测试工作流或 Activity
- 02验证工作流确定性和历史兼容性
- 03为 Temporal 测试配置本地环境和 CI
安装
复制下面的命令,在终端中运行。
$
npx skills add https:/ / github.com/ wshobson/ agents --skill temporal-python-testing安装后在 Claude Code 中输入/temporal-python-testing确认已加载。
包含的文件
除 SKILL.md 外,安装时一起下载的参考资料和脚本。
resources/integration-testing.mdlocal-setup.mdreplay-testing.mdunit-testing.md详细介绍
Temporal Python 测试策略
使用 pytest 对 Temporal 工作流进行全面测试的方法,针对特定测试场景提供渐进式披露资源。
何时使用本技能
- 单元测试工作流 - 使用时间跳跃进行快速测试
- 集成测试 - 使用模拟活动测试工作流
- 重放测试 - 对照生产历史验证确定性
- 本地开发 - 设置 Temporal 服务与 pytest
- CI/CD 集成 - 自动化测试流水线
- 覆盖率策略 - 达到 ≥80% 的测试覆盖率
测试理念
推荐方法(来源:docs.temporal.io/develop/python/testing-suite):
- 将大部分测试编写为集成测试
- 使用带异步 fixture 的 pytest
- 时间跳跃可实现快速反馈(长达数月的工作流 → 几秒钟)
- 模拟活动以隔离工作流逻辑
- 通过重放测试验证确定性
三种测试类型:
- 单元:使用时间跳跃测试工作流,使用 ActivityEnvironment 测试活动
- 集成:使用模拟活动测试 Worker
- 端到端:使用真实活动运行完整的 Temporal 服务(谨慎使用)
可用资源
本技能通过渐进式披露提供详细指导。根据你的测试需求加载特定资源:
单元测试资源
文件:resources/unit-testing.md
何时加载:隔离测试单个工作流或活动
包含内容:
- 带时间跳跃的 WorkflowEnvironment
- 用于活动测试的 ActivityEnvironment
- 长时间运行工作流的快速执行
- 手动推进时间的模式
- pytest fixture 与模式
集成测试资源
文件:resources/integration-testing.md
何时加载:使用模拟的外部依赖测试工作流
包含内容:
- 活动模拟策略
- 错误注入模式
- 多活动工作流测试
- 信号与查询测试
- 覆盖率策略
重放测试资源
文件: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
附加参考资料
- Python SDK 测试:docs.temporal.io/develop/python/testing-suite
- 测试模式:github.com/temporalio/temporal/blob/main/docs/development/testing.md
- Python 示例:github.com/temporalio/samples-python
评论
支持 Markdown