temporal-python-testing

Temporal Python Testing Strategies

Comprehensive testing approaches for Temporal workflows using pytest, progressive disclosure resources for specific testing scenarios.

Do not use this skill when

• The task is unrelated to temporal python testing strategies
• You need a different domain or tool outside this scope

Instructions

• Clarify goals, constraints, and required inputs.
• Apply relevant best practices and validate outcomes.
• Provide actionable steps and verification.

Use this skill when

• 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 a project-defined threshold (80% here is illustrative) test coverage

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 (long timer waits can be accelerated; activities still consume real time)
• 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)

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

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

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

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

Runtime prerequisites

Pin compatible temporalio, pytest and pytest-asyncio versions in the target project; configure asyncio_mode = auto or use pytest_asyncio.fixture explicitly. Snippets using YourWorkflow, activities, expected values and imports are adaptation sketches, not a bundled runnable test suite. Define sandboxed workflows/activities at module scope in importable modules; nested classes/functions in snippets illustrate logic but can fail sandbox re-import validation. Use unique workflow IDs/task queues and bound execution time. Time-skipping starts/downloads a test server where supported; isolate it from production.

Quick Start Guide

Basic Workflow Test

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

Basic Activity Test

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

Illustrative project coverage target (not an official Temporal threshold):

• Workflows: a project-defined threshold (80% here is illustrative) logic coverage
• Activities: a project-defined threshold (80% here is illustrative) logic coverage
• Integration: Critical paths with mocked activities
• Replay: All workflow versions before deployment

Key Testing Principles

1. Time-Skipping - Timer skipping does not accelerate real activity I/O
2. Mock Activities - Isolate workflow logic from external dependencies
3. Replay Testing - Validate determinism before deployment
4. High Coverage - a project-defined threshold (80% here is illustrative) target for production workflows
5. Fast Feedback - Measure actual test duration; server startup and activities take real time

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

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

Limitations

• Use this skill only when the task clearly matches the scope described above.
• Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
• Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.