Master Temporal Python Testing Strategies

Temporal Python Testing Strategies covers comprehensive testing approaches for Temporal workflows using pytest, delivered through progressive-disclosure resource files loaded only for the specific scenario at hand. Recommended testing philosophy: write the majority as integration tests, use pytest with async fixtures, rely on time-skipping so month-long workflows execute in seconds, mock activities to isolate workflow logic, and validate determinism with replay testing before deployment.

It defines three test types: Unit tests (workflows tested with time-skipping, activities tested in isolation with ActivityEnvironment), Integration tests (workers running with mocked activities), and End-to-end tests (a full Temporal server with real activities, to be used sparingly). Four progressive-disclosure resource files, each loaded only when needed: resources/unit-testing.md for WorkflowEnvironment time-skipping and ActivityEnvironment patterns; resources/integration-testing.md for activity mocking, error injection, multi-activity and signal/query testing; resources/replay-testing.md for determinism validation against production histories and CI/CD replay integration; and resources/local-setup.md for Docker Compose, pytest configuration and coverage tooling.

A basic workflow test starts a time-skipping environment as a pytest fixture, runs a Worker against it, and executes the workflow to assert on its result:

@pytest.fixture
async def workflow_env():
    env = await WorkflowEnvironment.start_time_skipping()
    yield env
    await env.shutdown()

Time-skipping means a workflow.sleep(timedelta(hours=24)) inside a workflow under test completes instantly rather than actually waiting, and the environment supports both automatic skipping and manual advancement via workflow_env.sleep(...) for precise control, including querying intermediate workflow state between advances. ActivityEnvironment tests activities without any workflow or Temporal server at all, and covers heartbeat reporting during long-running activities, cancellation handling via activity.is_cancelled(), and exception propagation - asserting a non-retryable ApplicationError is raised with the expected message. Integration-testing patterns mock an activity to test workflow orchestration logic in isolation, use dynamic mock functions that branch on input to exercise both success and error paths, and inject transient failures to verify a configured RetryPolicy actually retries (checking both the final result and the attempt count) while confirming a non-retryable ApplicationError short-circuits retries entirely for validation failures.

Coverage targets: workflows and activities should each reach at least 80% logic coverage, integration tests should cover critical paths with mocked activities, and replay testing should run against every workflow version before deployment. Pytest integration patterns include module-scoped fixtures to reuse a WorkflowEnvironment across related tests, function-scoped ActivityEnvironment fixtures for isolation, and @pytest.mark.parametrize for exercising multiple input/expected-output pairs against one activity test.

Replay testing catches non-deterministic workflow code that would otherwise break on a real Temporal replay. The named non-deterministic patterns and their fixes: calling random.randint() directly (different on every replay) instead of the deterministic workflow.random(); reading datetime.now() directly instead of the deterministic workflow.now(); and making a direct external call such as requests.get(...) from inside workflow code instead of routing it through workflow.execute_activity(). Replay testing itself runs a Replayer against exported production workflow histories, and a GitHub Actions example wires this into CI so workflow code changes are replay-tested before merge. Version compatibility testing uses workflow.get_version("feature-flag", 1, 2) to branch safely between old and new workflow behavior for code already running in production, and the same pattern anchors a two-phase migration strategy: add the version check so old and new logic coexist, then remove the old branch once every old-version workflow execution has completed. Common replay errors: a WorkflowNonDeterministicError (a command mismatch between the recorded history and the current code) means a code change altered the decision sequence, and a WorkflowVersionError (a workflow version changed without using get_version()) is fixed by wrapping the change in workflow.get_version(). Local development uses a Docker Compose file to run a Temporal server and its Web UI, plus a pytest.ini with test-category markers, coverage settings and an async test timeout, and a Makefile for common tasks like starting the server, running tests with coverage, and stopping the server.

Use it for unit testing workflows with time-skipping, integration testing with mocked activities, replay testing to validate determinism against production histories, local Temporal-server and pytest setup, CI/CD test pipeline integration, or working toward an 80%+ coverage target. Do not use it for tasks unrelated to Temporal Python testing strategies, or when a different domain or tool fits better.

Input is the Temporal workflow or activity code under test. Output is a pytest test suite organized by the three test types, using WorkflowEnvironment/ActivityEnvironment fixtures, with coverage reports validated against the 80% workflow/activity target and, before deployment, a replay-tested determinism check.

It integrates with pytest and pytest-asyncio, Temporal's Python SDK testing utilities (temporalio.testing.WorkflowEnvironment, ActivityEnvironment), Docker Compose for local Temporal server setup, and CI/CD pipelines for automated replay and coverage checks.

It suits Python developers building Temporal workflows who need fast, deterministic tests via time-skipping, isolated activity tests, mocked-dependency integration tests, and a replay-testing gate before shipping workflow changes to production.