Temporal.io

Temporal.io Testing Guide: Unit Testing Workflows and Activities

HelpMeTest

19 May 2026 — 5 min read

Temporal.io solves distributed workflow reliability by making long-running processes durable, resumable, and observable. The tradeoff is a testing model that differs significantly from standard service testing. Workflows must be deterministic — the same history must always produce the same decisions — and Activities are where all the non-deterministic work happens.

This guide covers the complete testing strategy for Temporal applications: unit testing with TestWorkflowEnvironment, activity testing, history replay for regression detection, and integration tests against a real Temporal server.

Understanding the Temporal Testing Model

Before writing tests, understand what Temporal's execution model means for testing:

Workflows are deterministic state machines. They schedule activities, handle signals, and manage timers — but they must never make non-deterministic calls (random numbers, current time, external I/O) directly. Temporal replays workflow history to reconstruct state, so any non-determinism causes "non-deterministic error" panics.

Activities are where you make external calls: database writes, API calls, file I/O. Activities are not required to be deterministic. They can fail and be retried automatically by Temporal.

This separation creates a clean testing boundary:

Test workflows with mocked activities using TestWorkflowEnvironment
Test activities as regular functions with real dependencies or mocks
Test the integration with a running Temporal server

Setting Up TestWorkflowEnvironment (Go)

// order_workflow_test.go
package workflows_test

import (
    "testing"
    "time"
    
    "github.com/stretchr/testify/mock"
    "github.com/stretchr/testify/require"
    "go.temporal.io/sdk/testsuite"
    "go.temporal.io/sdk/temporal"
    
    "myapp/workflows"
    "myapp/activities"
)

type OrderWorkflowTestSuite struct {
    suite.Suite
    testsuite.WorkflowTestSuite
    env *testsuite.TestWorkflowEnvironment
}

func (s *OrderWorkflowTestSuite) SetupTest() {
    s.env = s.NewTestWorkflowEnvironment()
}

func (s *OrderWorkflowTestSuite) AfterTest(suiteName, testName string) {
    s.env.AssertExpectations(s.T())
}

func TestOrderWorkflow(t *testing.T) {
    suite.Run(t, new(OrderWorkflowTestSuite))
}

Testing Workflow Happy Path

func (s *OrderWorkflowTestSuite) Test_OrderWorkflow_HappyPath() {
    order := workflows.Order{
        ID:       "order-123",
        Items:    []string{"widget-a", "widget-b"},
        Customer: "customer-456",
    }
    
    // Mock each activity the workflow calls
    s.env.OnActivity(activities.ValidateInventory, mock.Anything, order.Items).
        Return(nil).
        Once()
    
    s.env.OnActivity(activities.ChargeCustomer, mock.Anything, order.Customer, mock.AnythingOfType("float64")).
        Return(&activities.ChargeResult{TransactionID: "txn-789"}, nil).
        Once()
    
    s.env.OnActivity(activities.FulfillOrder, mock.Anything, mock.Anything).
        Return(nil).
        Once()
    
    s.env.OnActivity(activities.SendConfirmationEmail, mock.Anything, mock.Anything).
        Return(nil).
        Once()
    
    // Execute the workflow
    s.env.ExecuteWorkflow(workflows.OrderWorkflow, order)
    
    require.True(s.T(), s.env.IsWorkflowCompleted())
    require.NoError(s.T(), s.env.GetWorkflowError())
    
    var result workflows.OrderResult
    require.NoError(s.T(), s.env.GetWorkflowResult(&result))
    require.Equal(s.T(), "completed", result.Status)
    require.Equal(s.T(), "txn-789", result.TransactionID)
}

func (s *OrderWorkflowTestSuite) Test_OrderWorkflow_InventoryFailed_CancelsOrder() {
    order := workflows.Order{ID: "order-999", Items: []string{"out-of-stock-item"}}
    
    // Inventory check fails — workflow should compensate
    s.env.OnActivity(activities.ValidateInventory, mock.Anything, order.Items).
        Return(temporal.NewApplicationError("Item out of stock", "OUT_OF_STOCK")).
        Once()
    
    // Charge should NOT be called if inventory fails
    s.env.OnActivity(activities.NotifyCustomerUnavailable, mock.Anything, mock.Anything).
        Return(nil).
        Once()
    
    s.env.ExecuteWorkflow(workflows.OrderWorkflow, order)
    
    require.True(s.T(), s.env.IsWorkflowCompleted())
    
    var result workflows.OrderResult
    s.env.GetWorkflowResult(&result)
    require.Equal(s.T(), "cancelled", result.Status)
    require.Equal(s.T(), "OUT_OF_STOCK", result.CancellationReason)
}

Testing Signals and Timers

Temporal supports signals (external events sent to running workflows) and timers. Both can be tested with TestWorkflowEnvironment:

func (s *OrderWorkflowTestSuite) Test_OrderWorkflow_CancellationSignal() {
    order := workflows.Order{ID: "order-cancel-test", Items: []string{"widget-a"}}
    
    s.env.OnActivity(activities.ValidateInventory, mock.Anything, mock.Anything).
        Return(nil).
        Once()
    
    // Charge takes a while — send cancel signal before it completes
    s.env.OnActivity(activities.ChargeCustomer, mock.Anything, mock.Anything, mock.Anything).
        Return(nil, temporal.NewCanceledError()).
        Once()
    
    s.env.OnActivity(activities.RefundCustomer, mock.Anything, mock.Anything).
        Return(nil).
        Maybe() // May be called if charge partially succeeded
    
    // Register signal sender — fires after 2 simulated seconds
    s.env.RegisterDelayedCallback(2*time.Second, func() {
        s.env.SignalWorkflow("cancel-order", nil)
    })
    
    s.env.ExecuteWorkflow(workflows.OrderWorkflow, order)
    
    require.True(s.T(), s.env.IsWorkflowCompleted())
    
    var result workflows.OrderResult
    s.env.GetWorkflowResult(&result)
    require.Equal(s.T(), "cancelled", result.Status)
}

func (s *OrderWorkflowTestSuite) Test_OrderWorkflow_Timeout() {
    // TestWorkflowEnvironment fast-forwards timers — no real waiting
    order := workflows.Order{ID: "order-timeout-test", Items: []string{"widget-a"}}
    
    s.env.OnActivity(activities.ValidateInventory, mock.Anything, mock.Anything).
        Return(nil).
        Once()
    
    // Simulate charge hanging — will hit the workflow's 30-minute timeout
    s.env.OnActivity(activities.ChargeCustomer, mock.Anything, mock.Anything, mock.Anything).
        After(31 * time.Minute). // Exceeds timeout
        Return(nil, temporal.NewTimeoutError(enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE, nil)).
        Once()
    
    s.env.ExecuteWorkflow(workflows.OrderWorkflow, order)
    
    require.True(s.T(), s.env.IsWorkflowCompleted())
    require.Error(s.T(), s.env.GetWorkflowError())
}

Testing Activities

Activities are regular functions — test them independently with real or mocked dependencies:

// activities/charge_customer_test.go
package activities_test

import (
    "context"
    "testing"
    
    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"
    "go.temporal.io/sdk/activity"
    "go.temporal.io/sdk/testsuite"
    
    "myapp/activities"
)

func TestChargeCustomer_Success(t *testing.T) {
    testSuite := &testsuite.WorkflowTestSuite{}
    env := testSuite.NewTestActivityEnvironment()
    
    // Register the activity
    env.RegisterActivity(&activities.Activities{
        PaymentClient: &mockPaymentClient{
            chargeResult: &PaymentResult{TransactionID: "txn-001", Success: true},
        },
    })
    
    val, err := env.ExecuteActivity(activities.ChargeCustomer, "customer-123", 99.99)
    
    require.NoError(t, err)
    
    var result activities.ChargeResult
    require.NoError(t, val.Get(&result))
    assert.Equal(t, "txn-001", result.TransactionID)
}

func TestChargeCustomer_RetryableError(t *testing.T) {
    // Verify that network errors are returned as retryable (not ApplicationError)
    env := newTestEnv()
    env.RegisterActivity(&activities.Activities{
        PaymentClient: &mockPaymentClient{
            err: &net.OpError{Op: "connect", Err: errors.New("connection refused")},
        },
    })
    
    _, err := env.ExecuteActivity(activities.ChargeCustomer, "customer-123", 99.99)
    
    // Should be retryable — not an application error
    var appErr *temporal.ApplicationError
    require.False(t, errors.As(err, &appErr) && appErr.NonRetryable(),
        "Network errors should be retryable")
}

History Replay for Regression Testing

Temporal's history replay lets you catch non-determinism bugs. Capture a workflow execution history, then replay it against your updated code:

// tests/replay/replay_test.go
package replay_test

import (
    "os"
    "testing"
    
    "go.temporal.io/sdk/worker"
    "myapp/workflows"
)

func TestWorkflowHistoryReplay(t *testing.T) {
    // Histories captured from production: golden files
    historyFiles, err := filepath.Glob("testdata/histories/*.json")
    require.NoError(t, err)
    require.NotEmpty(t, historyFiles, "No history files found in testdata/histories/")
    
    replayer, err := worker.NewWorkflowReplayer()
    require.NoError(t, err)
    
    replayer.RegisterWorkflow(workflows.OrderWorkflow)
    
    for _, historyFile := range historyFiles {
        t.Run(filepath.Base(historyFile), func(t *testing.T) {
            err := replayer.ReplayWorkflowHistoryFromJSONFile(nil, historyFile)
            require.NoError(t, err, "Non-determinism detected in %s", historyFile)
        })
    }
}

Capture histories from your Temporal server:

# Capture a workflow history for replay testing
temporal workflow show \
  --workflow-id order-123 \
  --run-id <run-id> \
  --output json > tests/replay/testdata/histories/order_happy_path.json

Python SDK: Using TestWorkflowEnvironment

The Python SDK provides equivalent test utilities:

# tests/test_order_workflow.py
import asyncio
import pytest
from unittest.mock import AsyncMock
from temporalio import activity
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

from myapp.workflows import OrderWorkflow, OrderInput
from myapp.activities import validate_inventory, charge_customer, fulfill_order

@pytest.mark.asyncio
async def test_order_workflow_happy_path():
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client,
            task_queue="test-queue",
            workflows=[OrderWorkflow],
            activities=[validate_inventory, charge_customer, fulfill_order],
        ):
            # Mock activities via patching
            validate_inventory_mock = AsyncMock(return_value=None)
            charge_customer_mock = AsyncMock(return_value={"transaction_id": "txn-001"})
            fulfill_order_mock = AsyncMock(return_value=None)
            
            result = await env.client.execute_workflow(
                OrderWorkflow.run,
                OrderInput(order_id="order-123", items=["widget-a"]),
                id="test-order-123",
                task_queue="test-queue",
            )
            
            assert result["status"] == "completed"
            assert result["transaction_id"] == "txn-001"

CI Integration

# .github/workflows/temporal-tests.yml
name: Temporal Workflow Tests

on: [push, pull_request]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with:
          go-version: "1.22"
      - name: Run workflow unit tests
        run: go test ./workflows/... ./activities/... -v -timeout 60s

  replay-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
      - name: Run history replay tests
        run: go test ./tests/replay/... -v

  integration-tests:
    runs-on: ubuntu-latest
    services:
      temporal:
        image: temporalio/auto-setup:latest
        ports:
          - 7233:7233
    steps:
      - uses: actions/checkout@v4
      - run: go test ./tests/integration/... -v -timeout 300s
        env:
          TEMPORAL_HOST: localhost:7233

Production Monitoring

Temporal's visibility API provides workflow status, but it doesn't alert you when key business workflows start failing at higher rates than expected. HelpMeTest can run scheduled tests that trigger real workflow executions and assert on completion — catching infrastructure regressions before they become customer-facing incidents.

Conclusion

Temporal testing has three layers: TestWorkflowEnvironment for fast, deterministic workflow logic tests; activity unit tests for your external integrations; and history replay tests to catch non-determinism regressions after refactors. The time-skipping feature of TestWorkflowEnvironment means timer-heavy workflows that would take hours in production run in milliseconds in tests. Capture golden histories from production and replay them in CI — that's your strongest safety net against breaking existing workflow clients.