AutoGineer
Generated banner for Temporal Python Testing
Generated logo for Temporal Python Testing

Temporal Python Testing

Seth HobsonCommunityCreated by Seth Hobson

  • Skill
  • Guidelines
  • Software development
  • MIT
  • Low risk
Download ZIP

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

  1. Download the skill as a ZIP:

    https://autogineer.com/agent-skills/wshobson/temporal-python-testing/download

  2. 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:

  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

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

  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

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

resources/local-setup.md

Local Development Setup for Temporal Python Testing

Comprehensive guide for setting up local Temporal development environment with pytest integration and coverage tracking.

Temporal Server Setup with Docker Compose

Basic Docker Compose Configuration
# docker-compose.yml
version: "3.8"

services:
  temporal:
    image: temporalio/auto-setup:1.26
    container_name: temporal-dev
    ports:
      - "7233:7233" # Temporal server
      - "8233:8233" # Web UI
    environment:
      - DB=postgresql
      - POSTGRES_USER=temporal
      - POSTGRES_PWD=temporal
      - POSTGRES_SEEDS=postgresql
      - DYNAMIC_CONFIG_FILE_PATH=config/dynamicconfig/development-sql.yaml
    depends_on:
      - postgresql

  postgresql:
    image: postgres:14-alpine
    container_name: temporal-postgres
    environment:
      - POSTGRES_USER=temporal
      - POSTGRES_PASSWORD=temporal
      - POSTGRES_DB=temporal
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  temporal-ui:
    image: temporalio/ui:2.33
    container_name: temporal-ui
    depends_on:
      - temporal
    environment:
      - TEMPORAL_ADDRESS=temporal:7233
      - TEMPORAL_CORS_ORIGINS=http://localhost:3000
    ports:
      - "8080:8080"

volumes:
  postgres_data:
Starting Local Server
# Start Temporal server
docker-compose up -d

# Verify server is running
docker-compose ps

# View logs
docker-compose logs -f temporal

# Access Temporal Web UI
open http://localhost:8080

# Stop server
docker-compose down

# Reset data (clean slate)
docker-compose down -v
Health Check Script
# scripts/health_check.py
import asyncio
from temporalio.client import Client

async def check_temporal_health():
    """Verify Temporal server is accessible"""
    try:
        client = await Client.connect("localhost:7233")
        print("✓ Connected to Temporal server")

        # Test workflow execution
        from temporalio.worker import Worker

        @workflow.defn
        class HealthCheckWorkflow:
            @workflow.run
            async def run(self) -> str:
                return "healthy"

        async with Worker(
            client,
            task_queue="health-check",
            workflows=[HealthCheckWorkflow],
        ):
            result = await client.execute_workflow(
                HealthCheckWorkflow.run,
                id="health-check",
                task_queue="health-check",
            )
            print(f"✓ Workflow execution successful: {result}")

        return True

    except Exception as e:
        print(f"✗ Health check failed: {e}")
        return False

if __name__ == "__main__":
    asyncio.run(check_temporal_health())

pytest Configuration

Project Structure
temporal-project/
├── docker-compose.yml
├── pyproject.toml
├── pytest.ini
├── requirements.txt
├── src/
│   ├── workflows/
│   │   ├── __init__.py
│   │   ├── order_workflow.py
│   │   └── payment_workflow.py
│   └── activities/
│       ├── __init__.py
│       ├── payment_activities.py
│       └── inventory_activities.py
├── tests/
│   ├── conftest.py
│   ├── unit/
│   │   ├── test_workflows.py
│   │   └── test_activities.py
│   ├── integration/
│   │   └── test_order_flow.py
│   └── replay/
│       └── test_workflow_replay.py
└── scripts/
    ├── health_check.py
    └── export_histories.py
pytest Configuration
# pytest.ini
[pytest]
asyncio_mode = auto
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# Markers for test categorization
markers =
    unit: Unit tests (fast, isolated)
    integration: Integration tests (require Temporal server)
    replay: Replay tests (require production histories)
    slow: Slow running tests

# Coverage settings
addopts =
    --verbose
    --strict-markers
    --cov=src
    --cov-report=term-missing
    --cov-report=html
    --cov-fail-under=80

# Async test timeout
asyncio_default_fixture_loop_scope = function
Shared Test Fixtures
# tests/conftest.py
import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.client import Client

@pytest.fixture(scope="session")
def event_loop():
    """Provide event loop for async fixtures"""
    import asyncio
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def temporal_client():
    """Provide Temporal client connected to local server"""
    client = await Client.connect("localhost:7233")
    yield client
    await client.close()

@pytest.fixture(scope="module")
async def workflow_env():
    """Module-scoped time-skipping environment"""
    env = await WorkflowEnvironment.start_time_skipping()
    yield env
    await env.shutdown()

@pytest.fixture
def activity_env():
    """Function-scoped activity environment"""
    from temporalio.testing import ActivityEnvironment
    return ActivityEnvironment()

@pytest.fixture
async def test_worker(temporal_client, workflow_env):
    """Pre-configured test worker"""
    from temporalio.worker import Worker
    from src.workflows import OrderWorkflow, PaymentWorkflow
    from src.activities import process_payment, update_inventory

    return Worker(
        workflow_env.client,
        task_queue="test-queue",
        workflows=[OrderWorkflow, PaymentWorkflow],
        activities=[process_payment, update_inventory],
    )
Dependencies
# requirements.txt
temporalio>=1.5.0
pytest>=7.4.0
pytest-asyncio>=0.21.0
pytest-cov>=4.1.0
pytest-xdist>=3.3.0  # Parallel test execution
# pyproject.toml
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_backend"

[project]
name = "temporal-project"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
    "temporalio>=1.5.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.4.0",
    "pytest-asyncio>=0.21.0",
    "pytest-cov>=4.1.0",
    "pytest-xdist>=3.3.0",
]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

Coverage Configuration

Coverage Settings
# .coveragerc
[run]
source = src
omit =
    */tests/*
    */venv/*
    */__pycache__/*

[report]
exclude_lines =
    # Exclude type checking blocks
    if TYPE_CHECKING:
    # Exclude debug code
    def __repr__
    # Exclude abstract methods
    @abstractmethod
    # Exclude pass statements
    pass

[html]
directory = htmlcov
Running Tests with Coverage
# Run all tests with coverage
pytest --cov=src --cov-report=term-missing

# Generate HTML coverage report
pytest --cov=src --cov-report=html
open htmlcov/index.html

# Run specific test categories
pytest -m unit  # Unit tests only
pytest -m integration  # Integration tests only
pytest -m "not slow"  # Skip slow tests

# Parallel execution (faster)
pytest -n auto  # Use all CPU cores

# Fail if coverage below threshold
pytest --cov=src --cov-fail-under=80
Coverage Report Example
---------- coverage: platform darwin, python 3.11.5 -----------
Name                                Stmts   Miss  Cover   Missing
-----------------------------------------------------------------
src/__init__.py                         0      0   100%
src/activities/__init__.py              2      0   100%
src/activities/inventory.py            45      3    93%   78-80
src/activities/payment.py              38      0   100%
src/workflows/__init__.py               2      0   100%
src/workflows/order_workflow.py        67      5    93%   45-49
src/workflows/payment_workflow.py      52      0   100%
-----------------------------------------------------------------
TOTAL                                 206      8    96%

10 files skipped due to complete coverage.

Development Workflow

Daily Development Flow
# 1. Start Temporal server
docker-compose up -d

# 2. Verify server health
python scripts/health_check.py

# 3. Run tests during development
pytest tests/unit/ --verbose

# 4. Run full test suite before commit
pytest --cov=src --cov-report=term-missing

# 5. Check coverage
open htmlcov/index.html

# 6. Stop server
docker-compose down
Pre-Commit Hook
# .git/hooks/pre-commit
#!/bin/bash

echo "Running tests..."
pytest --cov=src --cov-fail-under=80

if [ $? -ne 0 ]; then
    echo "Tests failed. Commit aborted."
    exit 1
fi

echo "All tests passed!"
Makefile for Common Tasks
# Makefile
.PHONY: setup test test-unit test-integration coverage clean

setup:
	docker-compose up -d
	pip install -r requirements.txt
	python scripts/health_check.py

test:
	pytest --cov=src --cov-report=term-missing

test-unit:
	pytest -m unit --verbose

test-integration:
	pytest -m integration --verbose

test-replay:
	pytest -m replay --verbose

test-parallel:
	pytest -n auto --cov=src

coverage:
	pytest --cov=src --cov-report=html
	open htmlcov/index.html

clean:
	docker-compose down -v
	rm -rf .pytest_cache htmlcov .coverage

ci:
	docker-compose up -d
	sleep 10  # Wait for Temporal to start
	pytest --cov=src --cov-fail-under=80
	docker-compose down
CI/CD Example
# .github/workflows/test.yml
name: Tests

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"

      - name: Start Temporal server
        run: docker-compose up -d

      - name: Wait for Temporal
        run: sleep 10

      - name: Install dependencies
        run: |
          pip install -r requirements.txt

      - name: Run tests with coverage
        run: |
          pytest --cov=src --cov-report=xml --cov-fail-under=80

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml

      - name: Cleanup
        if: always()
        run: docker-compose down

Debugging Tips

Enable Temporal SDK Logging
import logging

# Enable debug logging for Temporal SDK
logging.basicConfig(level=logging.DEBUG)
temporal_logger = logging.getLogger("temporalio")
temporal_logger.setLevel(logging.DEBUG)
Interactive Debugging
# Add breakpoint in test
@pytest.mark.asyncio
async def test_workflow_with_breakpoint(workflow_env):
    import pdb; pdb.set_trace()  # Debug here

    async with Worker(...):
        result = await workflow_env.client.execute_workflow(...)
Temporal Web UI
# Access Web UI at http://localhost:8080
# - View workflow executions
# - Inspect event history
# - Replay workflows
# - Monitor workers

Best Practices

  1. Isolated Environment: Use Docker Compose for reproducible local setup

  2. Health Checks: Always verify Temporal server before running tests

  3. Fast Feedback: Use pytest markers to run unit tests quickly

  4. Coverage Targets: Maintain ≥80% code coverage

  5. Parallel Testing: Use pytest-xdist for faster test runs

  6. CI/CD Integration: Automated testing on every commit

  7. Cleanup: Clear Docker volumes between test runs if needed

Troubleshooting

Issue: Temporal server not starting

# Check logs
docker-compose logs temporal

# Reset database
docker-compose down -v
docker-compose up -d

Issue: Tests timing out

# Increase timeout in pytest.ini
asyncio_default_timeout = 30

Issue: Port already in use

# Find process using port 7233
lsof -i :7233

# Kill process or change port in docker-compose.yml

Additional Resources

  • Temporal Local Development: docs.temporal.io/develop/python/local-dev

  • pytest Documentation: docs.pytest.org

  • Docker Compose: docs.docker.com/compose

  • pytest-asyncio: github.com/pytest-dev/pytest-asyncio

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

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