Back to Details

temporal-python-testing

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Temporal Python Testing Strategies

Temporal Python 测试策略

Comprehensive testing approaches for Temporal workflows using pytest, progressive disclosure resources for specific testing scenarios.
使用pytest进行Temporal工作流的全面测试方法，针对特定测试场景提供渐进式参考资源。

When to Use This Skill

何时使用该技能

  • Unit testing workflows - Fast tests with time-skipping
  • Integration testing - Workflows with mocked activities
  • Replay testing - Validate determinism against production histories
  • Local development - Set up Temporal server and pytest
  • CI/CD integration - Automated testing pipelines
  • Coverage strategies - Achieve ≥80% test coverage
  • 单元测试工作流 - 支持时间跳过的快速测试
  • 集成测试 - 包含模拟活动的工作流测试
  • 重放测试 - 根据生产历史验证确定性
  • 本地开发 - 搭建Temporal服务器和pytest环境
  • CI/CD集成 - 自动化测试流水线
  • 覆盖率策略 - 实现≥80%的测试覆盖率

Testing Philosophy

测试理念

Recommended Approach (Source: docs.temporal.io/develop/python/testing-suite):
  • Write majority as integration tests
  • Use pytest with async fixtures
  • Time-skipping enables fast feedback (month-long workflows → seconds)
  • Mock activities to isolate workflow logic
  • Validate determinism with replay testing
Three Test Types:
  1. Unit: Workflows with time-skipping, activities with ActivityEnvironment
  2. Integration: Workers with mocked activities
  3. End-to-end: Full Temporal server with real activities (use sparingly)
推荐方法（来源：docs.temporal.io/develop/python/testing-suite）：
  • 大部分测试采用集成测试形式
  • 结合pytest与异步fixtures
  • 时间跳过实现快速反馈（将长达数月的工作流测试缩短至数秒）
  • 模拟活动以隔离工作流逻辑
  • 通过重放测试验证确定性
三种测试类型：
  1. 单元测试：带时间跳过的工作流，使用ActivityEnvironment测试活动
  2. 集成测试：包含模拟活动的Worker测试
  3. 端到端测试：搭配真实活动的完整Temporal服务器测试（谨慎使用）

Available Resources

可用资源

This skill provides detailed guidance through progressive disclosure. Load specific resources based on your testing needs:
本技能针对不同测试场景提供详细指导，可根据需求加载特定资源：

Unit Testing Resources

单元测试资源

File: 
resources/unit-testing.md
When to load: Testing individual workflows or activities in isolation Contains:
  • WorkflowEnvironment with time-skipping
  • ActivityEnvironment for activity testing
  • Fast execution of long-running workflows
  • Manual time advancement patterns
  • pytest fixtures and patterns
文件：
resources/unit-testing.md
加载时机：隔离测试单个工作流或活动时 包含内容：
  • 支持时间跳过的WorkflowEnvironment
  • 用于活动测试的ActivityEnvironment
  • 长运行工作流的快速执行
  • 手动时间推进模式
  • pytest fixtures及使用模式

Integration Testing Resources

集成测试资源

File: 
resources/integration-testing.md
When to load: Testing workflows with mocked external dependencies Contains:
  • Activity mocking strategies
  • Error injection patterns
  • Multi-activity workflow testing
  • Signal and query testing
  • Coverage strategies
文件：
resources/integration-testing.md
加载时机：测试包含模拟外部依赖的工作流时 包含内容：
  • 活动模拟策略
  • 错误注入模式
  • 多活动工作流测试
  • 信号与查询测试
  • 覆盖率策略

Replay Testing Resources

重放测试资源

File: 
resources/replay-testing.md
When to load: Validating determinism or deploying workflow changes Contains:
  • Determinism validation
  • Production history replay
  • CI/CD integration patterns
  • Version compatibility testing
文件：
resources/replay-testing.md
加载时机：验证确定性或部署工作流变更前 包含内容：
  • 确定性验证
  • 生产历史重放
  • CI/CD集成模式
  • 版本兼容性测试

Local Development Resources

本地开发资源

File: 
resources/local-setup.md
When to load: Setting up development environment Contains:
  • Docker Compose configuration
  • pytest setup and configuration
  • Coverage tool integration
  • Development workflow
文件：
resources/local-setup.md
加载时机：搭建开发环境时 包含内容：
  • Docker Compose配置
  • pytest环境搭建与配置
  • 覆盖率工具集成
  • 开发工作流

Quick Start Guide

快速入门指南

Basic Workflow Test

基础工作流测试

python
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
python
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

Basic Activity Test

基础活动测试

python
from temporalio.testing import ActivityEnvironment

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

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

Coverage Targets

覆盖率目标

Recommended Coverage (Source: docs.temporal.io best practices):
  • Workflows: ≥80% logic coverage
  • Activities: ≥80% logic coverage
  • Integration: Critical paths with mocked activities
  • Replay: All workflow versions before deployment
推荐覆盖率（来源：docs.temporal.io最佳实践）：
  • 工作流：≥80%逻辑覆盖率
  • 活动：≥80%逻辑覆盖率
  • 集成测试：针对关键路径使用模拟活动
  • 重放测试：部署前覆盖所有工作流版本

Key Testing Principles

核心测试原则

  1. Time-Skipping - Month-long workflows test in seconds
  2. Mock Activities - Isolate workflow logic from external dependencies
  3. Replay Testing - Validate determinism before deployment
  4. High Coverage - ≥80% target for production workflows
  5. Fast Feedback - Unit tests run in milliseconds
  1. 时间跳过 - 数月时长的工作流可在数秒内完成测试
  2. 模拟活动 - 将工作流逻辑与外部依赖隔离
  3. 重放测试 - 部署前验证确定性
  4. 高覆盖率 - 生产级工作流目标≥80%覆盖率
  5. 快速反馈 - 单元测试毫秒级运行

How to Use Resources

资源使用方式

Load specific resource when needed:
  • "Show me unit testing patterns" → Load 
    resources/unit-testing.md
  • "How do I mock activities?" → Load 
    resources/integration-testing.md
  • "Setup local Temporal server" → Load 
    resources/local-setup.md
  • "Validate determinism" → Load 
    resources/replay-testing.md
按需加载特定资源：
  • "展示单元测试模式" → 加载
    resources/unit-testing.md
  • "如何模拟活动？" → 加载
    resources/integration-testing.md
  • "搭建本地Temporal服务器" → 加载
    resources/local-setup.md
  • "验证确定性" → 加载
    resources/replay-testing.md

Additional References

额外参考

  • Python SDK Testing: docs.temporal.io/develop/python/testing-suite
  • Testing Patterns: github.com/temporalio/temporal/blob/main/docs/development/testing.md
  • Python Samples: github.com/temporalio/samples-python
  • 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