关于 temporal-python-testing
面向使用 Temporal Python SDK 的开发者,提供工作流与活动的测试策略及分场景参考资料,涵盖时间跳跃、模拟活动和回放测试。
主要功能
- 01工作流时间跳跃测试
- 02活动隔离测试
- 03集成测试与依赖模拟
- 04回放测试与确定性验证
- 05pytest 与本地开发配置
适用场景
- 01快速测试长时间运行的工作流
- 02隔离验证工作流逻辑
- 03在部署前检查确定性
- 04配置 Temporal 测试及 CI/CD
安装
复制下面的命令,在终端中运行。
$
npx skills add https:/ / github.com/ ormus-solutions/ libreuiux-claude-code --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
加载时机:测试带有模拟外部依赖项的工作流
包含内容:
- 活动模拟策略
- 错误注入模式
- 多活动工作流测试
- 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
其他参考资料
- 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