古法编程

如何使用 pytest 为 Temporal Python 工作流编写测试

解决分布式工作流测试难点：通过 AI 指导使用 pytest 和 Temporal 专用测试环境，实现快速的时间跳跃测试、Activity Mock 模拟以及生产历史重放验证。

为什么需要这个技能

Temporal 工作流通常涉及长达数天甚至数月的运行周期，且依赖于外部 Activity。如果直接在生产环境或实时环境中测试，将面临等待时间过长、外部依赖不可控以及难以验证“确定性（Determinism）”等问题。

本技能教你如何利用 WorkflowEnvironment 的时间跳跃（Time-skipping）特性将月级工作流在数秒内跑完，并通过 Mock 机制隔离外部依赖，确保工作流逻辑在各种边缘情况下都能正确运行。

适用场景

  • 单元测试：快速验证单个 Workflow 或 Activity 的逻辑，尤其是包含 sleep 的长时工作流。
  • 集成测试：在 Mock 外部 Activity 的情况下，测试工作流的整体编排逻辑。
  • 重放测试（Replay Testing）：在部署新版本前，使用生产环境的历史记录验证代码变更是否破坏了确定性。
  • CI/CD 集成：构建自动化测试流水线，确保代码覆盖率达到 80% 以上。

核心工作流

  1. 环境搭建：配置 pytest-asyncio 并使用 WorkflowEnvironment.start_time_skipping() 启动支持时间跳跃的测试环境。
  2. 逻辑隔离：
    • 使用 ActivityEnvironment 独立测试 Activity。
    • 在 Workflow 测试中使用 Mock 替代真实 Activity 接口。
  3. 确定性验证：通过重放测试，对比新旧代码在相同事件流下的状态转换是否一致。
  4. 覆盖率分析：针对 Workflow 编排路径和 Activity 业务逻辑分别进行覆盖率检查。

快速上手示例

基础工作流测试：

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

基础 Activity 测试：

from temporalio.testing import ActivityEnvironment

async def test_activity():
    env = ActivityEnvironment()
    result = await env.run(your_activity, "test-input")
    assert result == expected_output

下载和安装

下载 temporal-python-testing 中文版 Skill ZIP

解压后将目录放入你的 AI 工具 skills 文件夹，重启工具后即可使用。具体路径参考内附的 USAGE.zh.md。

你可能还需要

暂无推荐