temporal-python-testing

Temporal Python 测试策略

使用 pytest 对 Temporal 工作流进行全面测试，并通过渐进式披露资源了解特定测试场景。

不要在以下情况下使用此技能

• 任务与 Temporal Python 测试策略无关
• 你需要使用此范围之外的其他领域或工具

说明

• 明确目标、约束和所需输入。
• 应用相关最佳实践并验证结果。
• 提供可执行的步骤和验证方法。

在以下情况下使用此技能

• 工作流单元测试 - 通过时间跳跃进行快速测试
• 集成测试 - 测试带有模拟活动的工作流
• 回放测试 - 根据生产历史记录验证确定性
• 本地开发 - 设置 Temporal 服务和 pytest
• CI/CD 集成 - 自动化测试流水线
• 覆盖率策略 - 达到项目定义的测试覆盖率阈值（此处的 80% 仅作示例）

测试理念

推荐方法（来源：docs.temporal.io/develop/python/testing-suite）：

• 将大多数测试编写为集成测试
• 使用 pytest 和异步 fixture
• 时间跳跃可加快反馈速度（长时间计时器等待可以加速；活动仍会消耗实际时间）
• 使用模拟活动隔离工作流逻辑
• 通过回放测试验证确定性

三种测试类型：

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 设置和配置
• 覆盖率工具集成
• 开发工作流

运行时先决条件

在目标项目中固定兼容的 temporalio、pytest 和 pytest-asyncio 版本；配置 asyncio_mode = auto，或显式使用 pytest_asyncio.fixture。使用 YourWorkflow、活动、预期值和导入语句的代码片段只是适配示例，并非附带的可运行测试套件。应在可导入模块的模块级定义沙箱工作流和活动；片段中的嵌套类/函数用于演示逻辑，但可能无法通过沙箱重新导入验证。使用唯一的工作流 ID/任务队列，并限制执行时间。在支持的情况下，时间跳跃会启动/下载测试服务；请将其与生产环境隔离。

快速入门指南

基本工作流测试

import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

@pytest.fixture
async def workflow_env():
    env = await WorkflowEnvironment.start_time_skipping()
    try:
        yield env
    finally:
        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,
            "test-input", # or args=[arg1, arg2] for multiple arguments
            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

覆盖率目标

项目覆盖率目标示例（并非 Temporal 官方阈值）：

• 工作流：项目定义的逻辑覆盖率阈值（此处的 80% 仅作示例）
• 活动：项目定义的逻辑覆盖率阈值（此处的 80% 仅作示例）
• 集成测试：使用模拟活动覆盖关键路径
• 回放测试：部署前测试所有工作流版本

测试的关键原则

1. 跳过时间 - 跳过计时器不会加速实际活动 I/O
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

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为环境特定验证、测试或专家审查的替代品。
• 如果缺少必需的输入、权限、安全边界或成功标准，请停止并请求澄清。