mcpField
T
temporal-python-testing

temporal-python-testing

github.com/ormus-solutions测试与质量12377 次安装4 周前更新
代码仓库

关于 temporal-python-testing

面向使用 Temporal Python SDK 的开发者,提供工作流与活动的测试策略及分场景参考资料,涵盖时间跳跃、模拟活动和回放测试。

主要功能

  1. 01工作流时间跳跃测试
  2. 02活动隔离测试
  3. 03集成测试与依赖模拟
  4. 04回放测试与确定性验证
  5. 05pytest 与本地开发配置

适用场景

  1. 01快速测试长时间运行的工作流
  2. 02隔离验证工作流逻辑
  3. 03在部署前检查确定性
  4. 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.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 加载时机:测试带有模拟外部依赖项的工作流 包含内容:

  • 活动模拟策略
  • 错误注入模式
  • 多活动工作流测试
  • 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% 逻辑覆盖率
  • 集成测试:使用模拟活动覆盖关键路径
  • 回放测试:部署前测试所有工作流版本

核心测试原则

  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