Temporal Python Testing

Seth HobsonCommunityCreated by Seth Hobson

Skill
Guidelines
Software development
MIT
Low risk

From the author

Test Temporal workflows with pytest, time-skipping, and mocking strategies. Covers unit testing, integration testing, replay testing, and local development setup. Use when implementing Temporal workflow tests or debugging test failures.

— the description by Seth Hobson

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

This skill provides a structured approach to testing Temporal workflows within the Python ecosystem. It covers the full testing lifecycle, from unit testing individual components with time-skipping to integration testing with mocked activities and validating workflow determinism via replay testing. The content emphasizes fast feedback loops and high coverage targets, offering specific patterns for local development and CI/CD integration.

What it does

Provides patterns for unit testing workflows using time-skipping environments.
Explains how to mock activities for isolated integration testing.
Outlines methods for validating workflow determinism through replay testing.
Offers guidance on setting up local Temporal development environments.
Defines best practices for achieving high test coverage in Temporal applications.
Includes boilerplate code for pytest fixtures and activity testing.

When to use it

When implementing a new test suite for Temporal workflows.
When debugging non-deterministic workflow failures.
When setting up a local development environment for Temporal.
When preparing workflows for production deployment with replay validation.
When optimizing test execution time for long-running workflows.

Advantages

Enables testing of long-running workflows in seconds via time-skipping.
Promotes isolation of workflow logic from external dependencies.
Ensures production reliability through determinism validation.
Provides a clear path for achieving 80% test coverage.
Supports both rapid local development and automated CI/CD pipelines.

Usage examples

“How do I test a workflow that runs for a month?”: The skill explains how to use WorkflowEnvironment.start_time_skipping to advance time manually, allowing month-long workflows to complete in seconds during testing.
“How can I mock an activity in my workflow test?”: The skill provides integration testing patterns that demonstrate how to use workers with mocked activities to isolate the workflow logic.
“How do I ensure my workflow changes are deterministic?”: The skill guides you through replay testing, which validates your current workflow code against historical execution data to ensure determinism.

Install

Download the skill as a ZIP:
https://autogineer.com/agent-skills/wshobson/temporal-python-testing/download

Unzip it into the cross-client skills folder of your project:

mkdir -p .agents/skills && unzip temporal-python-testing.zip -d .agents/skills

Documentation by Seth Hobson

Temporal Python Testing Strategies

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

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

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:

Unit: Workflows with time-skipping, activities with ActivityEnvironment
Integration: Workers with mocked activities
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

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()
    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

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

Key Testing Principles

Time-Skipping - Month-long workflows test in seconds
Mock Activities - Isolate workflow logic from external dependencies
Replay Testing - Validate determinism before deployment
High Coverage - ≥80% target for production workflows
Fast Feedback - Unit tests run in milliseconds

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

Files

ATTRIBUTION.mdAdded by AutoGineer1007 B
LICENSELicence file added by AutoGineer1.0 KiB
SKILL.md4.8 KiB
resources/integration-testing.md12.7 KiB
resources/local-setup.md11.8 KiB
resources/replay-testing.md12.1 KiB
resources/unit-testing.md8.5 KiB

resources/integration-testing.md

Integration Testing with Mocked Activities

Comprehensive patterns for testing workflows with mocked external dependencies, error injection, and complex scenarios.

Activity Mocking Strategy

Purpose: Test workflow orchestration logic without calling real external services

Basic Mock Pattern

import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from unittest.mock import Mock

@pytest.mark.asyncio
async def test_workflow_with_mocked_activity(workflow_env):
    """Mock activity to test workflow logic"""

    # Create mock activity
    mock_activity = Mock(return_value="mocked-result")

    @workflow.defn
    class WorkflowWithActivity:
        @workflow.run
        async def run(self, input: str) -> str:
            result = await workflow.execute_activity(
                process_external_data,
                input,
                start_to_close_timeout=timedelta(seconds=10),
            )
            return f"processed: {result}"

    async with Worker(
        workflow_env.client,
        task_queue="test",
        workflows=[WorkflowWithActivity],
        activities=[mock_activity],  # Use mock instead of real activity
    ):
        result = await workflow_env.client.execute_workflow(
            WorkflowWithActivity.run,
            "test-input",
            id="wf-mock",
            task_queue="test",
        )
        assert result == "processed: mocked-result"
        mock_activity.assert_called_once()

Dynamic Mock Responses

Scenario-Based Mocking:

@pytest.mark.asyncio
async def test_workflow_multiple_mock_scenarios(workflow_env):
    """Test different workflow paths with dynamic mocks"""

    # Mock returns different values based on input
    def dynamic_activity(input: str) -> str:
        if input == "error-case":
            raise ApplicationError("Validation failed", non_retryable=True)
        return f"processed-{input}"

    @workflow.defn
    class DynamicWorkflow:
        @workflow.run
        async def run(self, input: str) -> str:
            try:
                result = await workflow.execute_activity(
                    dynamic_activity,
                    input,
                    start_to_close_timeout=timedelta(seconds=10),
                )
                return f"success: {result}"
            except ApplicationError as e:
                return f"error: {e.message}"

    async with Worker(
        workflow_env.client,
        task_queue="test",
        workflows=[DynamicWorkflow],
        activities=[dynamic_activity],
    ):
        # Test success path
        result_success = await workflow_env.client.execute_workflow(
            DynamicWorkflow.run,
            "valid-input",
            id="wf-success",
            task_queue="test",
        )
        assert result_success == "success: processed-valid-input"

        # Test error path
        result_error = await workflow_env.client.execute_workflow(
            DynamicWorkflow.run,
            "error-case",
            id="wf-error",
            task_queue="test",
        )
        assert "Validation failed" in result_error

Error Injection Patterns

Testing Transient Failures

Retry Behavior:

@pytest.mark.asyncio
async def test_workflow_transient_errors(workflow_env):
    """Test retry logic with controlled failures"""

    attempt_count = 0

    @activity.defn
    async def transient_activity() -> str:
        nonlocal attempt_count
        attempt_count += 1

        if attempt_count < 3:
            raise Exception(f"Transient error {attempt_count}")
        return "success-after-retries"

    @workflow.defn
    class RetryWorkflow:
        @workflow.run
        async def run(self) -> str:
            return await workflow.execute_activity(
                transient_activity,
                start_to_close_timeout=timedelta(seconds=10),
                retry_policy=RetryPolicy(
                    initial_interval=timedelta(milliseconds=10),
                    maximum_attempts=5,
                    backoff_coefficient=1.0,
                ),
            )

    async with Worker(
        workflow_env.client,
        task_queue="test",
        workflows=[RetryWorkflow],
        activities=[transient_activity],
    ):
        result = await workflow_env.client.execute_workflow(
            RetryWorkflow.run,
            id="retry-wf",
            task_queue="test",
        )
        assert result == "success-after-retries"
        assert attempt_count == 3

Testing Non-Retryable Errors

Business Validation Failures:

@pytest.mark.asyncio
async def test_workflow_non_retryable_error(workflow_env):
    """Test handling of permanent failures"""

    @activity.defn
    async def validation_activity(input: dict) -> str:
        if not input.get("valid"):
            raise ApplicationError(
                "Invalid input",
                non_retryable=True,  # Don't retry validation errors
            )
        return "validated"

    @workflow.defn
    class ValidationWorkflow:
        @workflow.run
        async def run(self, input: dict) -> str:
            try:
                return await workflow.execute_activity(
                    validation_activity,
                    input,
                    start_to_close_timeout=timedelta(seconds=10),
                )
            except ApplicationError as e:
                return f"validation-failed: {e.message}"

    async with Worker(
        workflow_env.client,
        task_queue="test",
        workflows=[ValidationWorkflow],
        activities=[validation_activity],
    ):
        result = await workflow_env.client.execute_workflow(
            ValidationWorkflow.run,
            {"valid": False},
            id="validation-wf",
            task_queue="test",
        )
        assert "validation-failed" in result

Multi-Activity Workflow Testing

Sequential Activity Pattern

@pytest.mark.asyncio
async def test_workflow_sequential_activities(workflow_env):
    """Test workflow orchestrating multiple activities"""

    activity_calls = []

    @activity.defn
    async def step_1(input: str) -> str:
        activity_calls.append("step_1")
        return f"{input}-step1"

    @activity.defn
    async def step_2(input: str) -> str:
        activity_calls.append("step_2")
        return f"{input}-step2"

    @activity.defn
    async def step_3(input: str) -> str:
        activity_calls.append("step_3")
        return f"{input}-step3"

    @workflow.defn
    class SequentialWorkflow:
        @workflow.run
        async def run(self, input: str) -> str:
            result_1 = await workflow.execute_activity(
                step_1,
                input,
                start_to_close_timeout=timedelta(seconds=10),
            )
            result_2 = await workflow.execute_activity(
                step_2,
                result_1,
                start_to_close_timeout=timedelta(seconds=10),
            )
            result_3 = await workflow.execute_activity(
                step_3,
                result_2,
                start_to_close_timeout=timedelta(seconds=10),
            )
            return result_3

    async with Worker(
        workflow_env.client,
        task_queue="test",
        workflows=[SequentialWorkflow],
        activities=[step_1, step_2, step_3],
    ):
        result = await workflow_env.client.execute_workflow(
            SequentialWorkflow.run,
            "start",
            id="seq-wf",
            task_queue="test",
        )
        assert result == "start-step1-step2-step3"
        assert activity_calls == ["step_1", "step_2", "step_3"]

Parallel Activity Pattern

@pytest.mark.asyncio
async def test_workflow_parallel_activities(workflow_env):
    """Test concurrent activity execution"""

    @activity.defn
    async def parallel_task(task_id: int) -> str:
        return f"task-{task_id}"

    @workflow.defn
    class ParallelWorkflow:
        @workflow.run
        async def run(self, task_count: int) -> list[str]:
            # Execute activities in parallel
            tasks = [
                workflow.execute_activity(
                    parallel_task,
                    i,
                    start_to_close_timeout=timedelta(seconds=10),
                )
                for i in range(task_count)
            ]
            return await asyncio.gather(*tasks)

    async with Worker(
        workflow_env.client,
        task_queue="test",
        workflows=[ParallelWorkflow],
        activities=[parallel_task],
    ):
        result = await workflow_env.client.execute_workflow(
            ParallelWorkflow.run,
            3,
            id="parallel-wf",
            task_queue="test",
        )
        assert result == ["task-0", "task-1", "task-2"]

Signal and Query Testing

Signal Handlers

@pytest.mark.asyncio
async def test_workflow_signals(workflow_env):
    """Test workflow signal handling"""

    @workflow.defn
    class SignalWorkflow:
        def __init__(self) -> None:
            self._status = "initialized"

        @workflow.run
        async def run(self) -> str:
            # Wait for completion signal
            await workflow.wait_condition(lambda: self._status == "completed")
            return self._status

        @workflow.signal
        async def update_status(self, new_status: str) -> None:
            self._status = new_status

        @workflow.query
        def get_status(self) -> str:
            return self._status

    async with Worker(
        workflow_env.client,
        task_queue="test",
        workflows=[SignalWorkflow],
    ):
        # Start workflow
        handle = await workflow_env.client.start_workflow(
            SignalWorkflow.run,
            id="signal-wf",
            task_queue="test",
        )

        # Verify initial state via query
        initial_status = await handle.query(SignalWorkflow.get_status)
        assert initial_status == "initialized"

        # Send signal
        await handle.signal(SignalWorkflow.update_status, "processing")

        # Verify updated state
        updated_status = await handle.query(SignalWorkflow.get_status)
        assert updated_status == "processing"

        # Complete workflow
        await handle.signal(SignalWorkflow.update_status, "completed")
        result = await handle.result()
        assert result == "completed"

Coverage Strategies

Workflow Logic Coverage

Target: ≥80% coverage of workflow decision logic

# Test all branches
@pytest.mark.parametrize("condition,expected", [
    (True, "branch-a"),
    (False, "branch-b"),
])
async def test_workflow_branches(workflow_env, condition, expected):
    """Ensure all code paths are tested"""
    # Test implementation
    pass

Activity Coverage

Target: ≥80% coverage of activity logic

# Test activity edge cases
@pytest.mark.parametrize("input,expected", [
    ("valid", "success"),
    ("", "empty-input-error"),
    (None, "null-input-error"),
])
async def test_activity_edge_cases(activity_env, input, expected):
    """Test activity error handling"""
    # Test implementation
    pass

Integration Test Organization

Test Structure

tests/
├── integration/
│   ├── conftest.py              # Shared fixtures
│   ├── test_order_workflow.py   # Order processing tests
│   ├── test_payment_workflow.py # Payment tests
│   └── test_fulfillment_workflow.py
├── unit/
│   ├── test_order_activities.py
│   └── test_payment_activities.py
└── fixtures/
    └── test_data.py             # Test data builders

Shared Fixtures

# conftest.py
import pytest
from temporalio.testing import WorkflowEnvironment

@pytest.fixture(scope="session")
async def workflow_env():
    """Session-scoped environment for integration tests"""
    env = await WorkflowEnvironment.start_time_skipping()
    yield env
    await env.shutdown()

@pytest.fixture
def mock_payment_service():
    """Mock external payment service"""
    return Mock()

@pytest.fixture
def mock_inventory_service():
    """Mock external inventory service"""
    return Mock()

Best Practices

Mock External Dependencies: Never call real APIs in tests
Test Error Scenarios: Verify compensation and retry logic
Parallel Testing: Use pytest-xdist for faster test runs
Isolated Tests: Each test should be independent
Clear Assertions: Verify both results and side effects
Coverage Target: ≥80% for critical workflows
Fast Execution: Use time-skipping, avoid real delays

Additional Resources

Mocking Strategies: docs.temporal.io/develop/python/testing-suite
pytest Best Practices: docs.pytest.org/en/stable/goodpractices.html
Python SDK Samples: github.com/temporalio/samples-python

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs Python

Security

What our scanner noticed· Low risk

Instructions that use the network

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Created by

Seth Hobson

Publisher

Seth Hobson (Community)

Source

wshobson/agents · plugins/backend-development/skills/temporal-python-testing · commit 4bde316

Licence

MIT License (redistribution allowed)

LICENSE (added by AutoGineer from LICENSE)
ATTRIBUTION.md (written by AutoGineer)

Synced

2026-11-09

AutoGineer is not affiliated with or endorsed by Seth Hobson or Seth Hobson.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download temporal-python-testing.zip