Temporal Python 测试策略

使用 pytest 对 Temporal 工作流进行综合测试的方法，通过渐进式披露机制，按具体测试场景提供针对性资源。

适用场景

• 工作流单元测试 —— 使用时间跳过（time-skipping）实现快速验证
• 集成测试 —— 工作流搭配模拟（mocked）的 activity 运行
• 重放测试（Replay testing） —— 基于生产环境历史记录验证确定性（determinism）
• 本地开发 —— 搭建 Temporal 服务端与 pytest 环境
• CI/CD 集成 —— 构建自动化测试流水线
• 覆盖率策略 —— 达成 ≥80% 的测试覆盖率目标

测试理念

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

• 以集成测试为主进行编写
• 结合 pytest 及异步 fixture 使用
• 利用时间跳过实现快速反馈（例如：运行时长为月级的工作流可在秒级完成测试）
• 对 activity 进行 mock，以隔离工作流核心逻辑
• 通过重放测试验证确定性

三类测试类型：

1. 单元测试：工作流使用时间跳过；activity 使用 ActivityEnvironment
2. 集成测试：Worker 启动并配合模拟的 activity 运行
3. 端到端测试：完整 Temporal 服务端 + 真实 activity（应谨慎使用）

可用资源

本技能采用渐进式披露方式提供详细指导。请根据实际测试需求加载对应资源：

单元测试资源

文件：resources/unit-testing.md
适用时机：独立测试单个工作流或 activity
内容包含：

• 支持时间跳过的 WorkflowEnvironment
• 用于 activity 测试的 ActivityEnvironment
• 长时间运行工作流的快速执行能力
• 手动推进时间的常用模式
• pytest fixture 与典型测试模式

集成测试资源

文件：resources/integration-testing.md
适用时机：测试依赖外部服务被模拟后的工作流
内容包含：

• activity 模拟策略
• 错误注入（error injection）模式
• 多 activity 工作流的测试方法
• Signal 与 Query 的测试方式
• 覆盖率提升策略

重放测试资源

文件：resources/replay-testing.md
适用时机：验证确定性，或在部署工作流变更前执行校验
内容包含：

• 确定性（determinism）验证方法
• 基于生产环境历史记录的重放流程
• CI/CD 中集成重放测试的实践模式
• 版本兼容性测试方案

本地开发资源

文件：resources/local-setup.md
适用时机：搭建本地开发环境
内容包含：

• Docker Compose 配置示例
• pytest 初始化与配置说明
• Coverage 工具集成方式
• 典型本地开发工作流

快速入门指南

基础工作流测试

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

覆盖率目标

推荐覆盖率标准（来源：docs.temporal.io 最佳实践）：

• 工作流：≥80% 的逻辑覆盖率
• activity：≥80% 的逻辑覆盖率
• 集成测试：覆盖关键路径，并使用模拟 activity
• 重放测试：所有待上线的工作流版本均需完成重放验证

核心测试原则

1. 时间跳过（Time-Skipping） —— 将月级工作流压缩至秒级完成测试
2. 模拟 activity（Mock Activities） —— 将工作流逻辑与外部依赖完全解耦
3. 重放测试（Replay Testing） —— 在部署前验证确定性
4. 高覆盖率（High Coverage） —— 生产环境工作流建议 ≥80%
5. 快速反馈（Fast Feedback） —— 单元测试应在毫秒级内完成

资源调用方式

按需加载指定资源：

• “展示单元测试模式” → 加载 resources/unit-testing.md
• “如何模拟 activity？” → 加载 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