mcpField
T
temporal-python-testing

temporal-python-testing

github.com/wshobson测试与质量41.9k11.1k 次安装2 周前更新
代码仓库

关于 temporal-python-testing

面向使用 Python SDK 开发 Temporal 工作流的开发者,提供基于 pytest 的分层测试策略,涵盖时间跳跃、模拟 Activity 和重放验证。

主要功能

  1. 01工作流与 Activity 单元测试
  2. 02时间跳跃测试长流程
  3. 03模拟 Activity 的集成测试
  4. 04重放测试与确定性验证
  5. 05本地开发和 CI/CD 配置指导

适用场景

  1. 01隔离测试工作流或 Activity
  2. 02验证工作流确定性和历史兼容性
  3. 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.md
local-setup.md
replay-testing.md
unit-testing.md

详细介绍

Temporal Python 测试策略

使用 pytest 对 Temporal 工作流进行全面测试的方法,针对特定测试场景提供渐进式披露资源。

何时使用本技能

  • 单元测试工作流 - 使用时间跳跃进行快速测试
  • 集成测试 - 使用模拟活动测试工作流
  • 重放测试 - 对照生产历史验证确定性
  • 本地开发 - 设置 Temporal 服务与 pytest
  • CI/CD 集成 - 自动化测试流水线
  • 覆盖率策略 - 达到 ≥80% 的测试覆盖率

测试理念

推荐方法(来源:docs.temporal.io/develop/python/testing-suite):

  • 将大部分测试编写为集成测试
  • 使用带异步 fixture 的 pytest
  • 时间跳跃可实现快速反馈(长达数月的工作流 → 几秒钟)
  • 模拟活动以隔离工作流逻辑
  • 通过重放测试验证确定性

三种测试类型:

  1. 单元:使用时间跳跃测试工作流,使用 ActivityEnvironment 测试活动
  2. 集成:使用模拟活动测试 Worker
  3. 端到端:使用真实活动运行完整的 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% 逻辑覆盖率
  • 集成:使用模拟活动覆盖关键路径
  • 重放:部署前覆盖所有工作流版本

关键测试原则

  1. 时间跳跃 - 长达数月的工作流可在几秒钟内完成测试
  2. 模拟活动 - 将工作流逻辑与外部依赖隔离
  3. 重放测试 - 部署前验证确定性
  4. 高覆盖率 - 生产工作流的目标为 ≥80%
  5. 快速反馈 - 单元测试在毫秒级内运行

如何使用资源

在需要时加载特定资源:

  • “向我展示单元测试模式” → 加载 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