Serverless

Step Functions & Workflow Testing: State Machine Test Strategies

HelpMeTest

HelpMeTest

— 8 min read
Step Functions & Workflow Testing: State Machine Test Strategies

AWS Step Functions lets you orchestrate Lambda functions, ECS tasks, and other AWS services into complex workflows. State machines are powerful, but they're also notoriously difficult to test — executions run asynchronously, state transitions depend on the output of one step feeding into the next, and error handling flows branch in ways that are easy to miss. A state machine that works in the happy path may silently fail at step 7 of a 10-step workflow when an upstream service returns an unexpected shape.

This guide covers a complete testing strategy for Step Functions: from testing individual Lambda steps in isolation, to verifying state machine logic locally with Step Functions Local, to running integration tests that confirm your full workflow behaves correctly.

The Testing Pyramid for Step Functions

Step Functions workflows have three natural layers to test:

  1. Unit tests for Lambda steps — test each Lambda handler in isolation, verifying it produces the correct output shape for given inputs
  2. State machine logic tests — verify that the state machine routes inputs to the right steps and handles errors and retries correctly, using Step Functions Local
  3. Integration tests — execute the full state machine against real AWS services (or LocalStack) and verify end-to-end behavior

Start at the bottom — most of your coverage should be unit tests on individual Lambda steps, which are fast and cheap. The state machine logic tests are where you catch routing bugs. Integration tests are the most expensive and should be kept minimal.

Unit Testing Individual Lambda Steps

Each Lambda step in a Step Functions workflow is just a Lambda function. Apply standard Lambda unit testing patterns.

Consider an order processing workflow:

# steps/validate_order.py
import json

def lambda_handler(event, context):
    """
    Input: {"orderId": "...", "userId": "...", "items": [...], "total": ...}
    Output: {"orderId": "...", "userId": "...", "items": [...], "total": ..., "validated": true}
    """
    order_id = event.get('orderId')
    user_id = event.get('userId')
    items = event.get('items', [])
    total = event.get('total', 0)

    errors = []

    if not order_id:
        errors.append('orderId is required')
    if not user_id:
        errors.append('userId is required')
    if not items:
        errors.append('Order must have at least one item')
    if total <= 0:
        errors.append('Order total must be positive')

    if errors:
        raise ValueError(f"Order validation failed: {'; '.join(errors)}")

    return {
        **event,
        'validated': True
    }

Unit tests:

# tests/unit/test_validate_order.py
import pytest
from steps.validate_order import lambda_handler

def make_valid_order():
    return {
        'orderId': 'order-123',
        'userId': 'user-456',
        'items': [{'sku': 'WIDGET-1', 'qty': 2, 'price': 9.99}],
        'total': 19.98
    }

def test_valid_order_passes_through():
    order = make_valid_order()
    result = lambda_handler(order, None)

    assert result['validated'] is True
    assert result['orderId'] == 'order-123'
    assert result['userId'] == 'user-456'
    assert result['items'] == order['items']

def test_valid_order_preserves_all_fields():
    order = make_valid_order()
    order['customField'] = 'extra-data'
    result = lambda_handler(order, None)

    assert result['customField'] == 'extra-data'

def test_missing_order_id_raises():
    order = make_valid_order()
    del order['orderId']

    with pytest.raises(ValueError, match='orderId is required'):
        lambda_handler(order, None)

def test_empty_items_raises():
    order = make_valid_order()
    order['items'] = []

    with pytest.raises(ValueError, match='at least one item'):
        lambda_handler(order, None)

def test_zero_total_raises():
    order = make_valid_order()
    order['total'] = 0

    with pytest.raises(ValueError, match='total must be positive'):
        lambda_handler(order, None)

Testing a step that calls AWS services:

# steps/charge_payment.py
import boto3
import json

def lambda_handler(event, context):
    """
    Calls external payment service and stores result in DynamoDB
    """
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('orders')

    order_id = event['orderId']
    total = event['total']

    # Call payment service (simplified)
    payment_result = call_payment_service(order_id, total)

    # Store result
    table.update_item(
        Key={'orderId': order_id},
        UpdateExpression='SET paymentStatus = :s, paymentId = :p',
        ExpressionAttributeValues={
            ':s': payment_result['status'],
            ':p': payment_result['paymentId']
        }
    )

    return {
        **event,
        'paymentStatus': payment_result['status'],
        'paymentId': payment_result['paymentId']
    }
# tests/unit/test_charge_payment.py
from unittest.mock import patch, MagicMock
from moto import mock_dynamodb
import boto3
import os

os.environ['AWS_DEFAULT_REGION'] = 'us-east-1'
os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'

@mock_dynamodb
def test_charge_payment_success():
    # Set up DynamoDB table
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.create_table(
        TableName='orders',
        KeySchema=[{'AttributeName': 'orderId', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'orderId', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )
    table.put_item(Item={'orderId': 'order-123', 'status': 'validated'})

    with patch('steps.charge_payment.call_payment_service') as mock_payment:
        mock_payment.return_value = {
            'status': 'charged',
            'paymentId': 'pay-abc123'
        }

        from steps.charge_payment import lambda_handler
        event = {'orderId': 'order-123', 'userId': 'user-456', 'total': 19.98}
        result = lambda_handler(event, None)

    assert result['paymentStatus'] == 'charged'
    assert result['paymentId'] == 'pay-abc123'

    # Verify DynamoDB was updated
    item = table.get_item(Key={'orderId': 'order-123'})['Item']
    assert item['paymentStatus'] == 'charged'

Testing State Machine Logic with Step Functions Local

Step Functions Local is a downloadable JAR that runs the state machine execution engine without connecting to AWS. It's the right tool for testing workflow routing, error handling, and retry behavior.

Download and start it:

# Download from AWS
curl -O https://s3.amazonaws.com/stepfunctionslocal/StepFunctionsLocal.tar.gz
tar -xf StepFunctionsLocal.tar.gz

# Start the local server
java -jar StepFunctionsLocal.jar --port 8083

Or with Docker:

docker run -p 8083:8083 amazon/aws-stepfunctions-local

Now create state machine definitions and execute them:

# tests/integration/test_state_machine.py
import json
import time
import boto3
import pytest

SF_LOCAL_ENDPOINT = 'http://localhost:8083'

@pytest.fixture(scope='session')
def sf_client():
    return boto3.client(
        'stepfunctions',
        endpoint_url=SF_LOCAL_ENDPOINT,
        region_name='us-east-1',
        aws_access_key_id='test',
        aws_secret_access_key='test'
    )

def load_definition(path):
    with open(path) as f:
        return f.read()

@pytest.fixture(scope='module')
def state_machine_arn(sf_client):
    definition = load_definition('statemachine/order-processing.asl.json')

    response = sf_client.create_state_machine(
        name='OrderProcessing-Test',
        definition=definition,
        roleArn='arn:aws:iam::123456789012:role/DummyRole'
    )
    arn = response['stateMachineArn']
    yield arn
    sf_client.delete_state_machine(stateMachineArn=arn)

def wait_for_execution(sf_client, execution_arn, timeout=30):
    """Poll until execution completes or timeout."""
    start = time.time()
    while time.time() - start < timeout:
        response = sf_client.describe_execution(executionArn=execution_arn)
        status = response['status']
        if status in ('SUCCEEDED', 'FAILED', 'TIMED_OUT', 'ABORTED'):
            return response
        time.sleep(0.5)
    raise TimeoutError(f"Execution {execution_arn} did not complete in {timeout}s")

def test_happy_path(sf_client, state_machine_arn):
    input_data = {
        'orderId': 'order-happy-1',
        'userId': 'user-456',
        'items': [{'sku': 'WIDGET-1', 'qty': 1, 'price': 9.99}],
        'total': 9.99
    }

    response = sf_client.start_execution(
        stateMachineArn=state_machine_arn,
        input=json.dumps(input_data)
    )
    execution_arn = response['executionArn']

    result = wait_for_execution(sf_client, execution_arn)

    assert result['status'] == 'SUCCEEDED'
    output = json.loads(result['output'])
    assert output['validated'] is True
    assert output['orderId'] == 'order-happy-1'

def test_validation_failure_routes_to_error_state(sf_client, state_machine_arn):
    # Send invalid input — missing userId
    input_data = {
        'orderId': 'order-invalid-1',
        'items': [{'sku': 'WIDGET-1', 'qty': 1}],
        'total': 9.99
    }

    response = sf_client.start_execution(
        stateMachineArn=state_machine_arn,
        input=json.dumps(input_data)
    )

    result = wait_for_execution(sf_client, response['executionArn'])

    # Should fail at validation step
    assert result['status'] == 'FAILED'
    assert 'userId is required' in result.get('cause', '')

State Machine Definition with Error Handling

{
  "Comment": "Order Processing Workflow",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:ValidateOrder",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException", "Lambda.TooManyRequestsException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3,
          "BackoffRate": 2
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["ValueError"],
          "Next": "OrderValidationFailed",
          "ResultPath": "$.error"
        }
      ],
      "Next": "ChargePayment"
    },
    "ChargePayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:ChargePayment",
      "Retry": [
        {
          "ErrorEquals": ["States.ALL"],
          "IntervalSeconds": 5,
          "MaxAttempts": 2,
          "BackoffRate": 1.5
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["PaymentDeclinedException"],
          "Next": "PaymentDeclined",
          "ResultPath": "$.error"
        }
      ],
      "Next": "FulfillOrder"
    },
    "FulfillOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:FulfillOrder",
      "Next": "OrderComplete"
    },
    "OrderComplete": {
      "Type": "Succeed"
    },
    "OrderValidationFailed": {
      "Type": "Fail",
      "Error": "ValidationError",
      "Cause": "Order failed validation"
    },
    "PaymentDeclined": {
      "Type": "Fail",
      "Error": "PaymentError",
      "Cause": "Payment was declined"
    }
  }
}

Mocking Lambda Steps in Step Functions Local

Step Functions Local can be configured to use mock Lambda responses instead of calling real Lambda functions. This lets you test state machine routing in isolation.

Create a MockConfigFile.json:

{
  "StateMachines": {
    "OrderProcessing-Test": {
      "TestCases": {
        "HappyPath": {
          "ValidateOrder": {
            "Return": {
              "orderId": "order-123",
              "userId": "user-456",
              "items": [{"sku": "WIDGET-1", "qty": 1}],
              "total": 9.99,
              "validated": true
            }
          },
          "ChargePayment": {
            "Return": {
              "orderId": "order-123",
              "paymentStatus": "charged",
              "paymentId": "pay-abc"
            }
          },
          "FulfillOrder": {
            "Return": {
              "orderId": "order-123",
              "status": "fulfilled"
            }
          }
        },
        "PaymentDeclined": {
          "ValidateOrder": {
            "Return": {
              "orderId": "order-123",
              "validated": true
            }
          },
          "ChargePayment": {
            "Throw": {
              "Error": "PaymentDeclinedException",
              "Cause": "Card declined"
            }
          }
        }
      }
    }
  }
}

Start Step Functions Local with the mock config:

java -jar StepFunctionsLocal.jar \
  --mock-config MockConfigFile.json \
  --lambda-endpoint http://localhost:3001

Then use the test case name in your execution:

def test_payment_declined_routing(sf_client, state_machine_arn):
    response = sf_client.start_execution(
        stateMachineArn=state_machine_arn,
        name='PaymentDeclinedTest',
        input=json.dumps({'orderId': 'order-123', 'total': 9.99})
    )

    result = wait_for_execution(sf_client, response['executionArn'])

    assert result['status'] == 'FAILED'
    # Verify it failed at the correct state
    history = sf_client.get_execution_history(
        executionArn=response['executionArn']
    )
    failed_events = [e for e in history['events'] if e['type'] == 'ExecutionFailed']
    assert len(failed_events) == 1

Testing with AWS CDK

If you use AWS CDK to define your state machines, you can write unit tests for the infrastructure definition itself using CDK's assertion library:

// test/order-processing-stack.test.ts
import { App } from 'aws-cdk-lib';
import { Template } from 'aws-cdk-lib/assertions';
import { OrderProcessingStack } from '../lib/order-processing-stack';

describe('OrderProcessingStack', () => {
    let template: Template;

    beforeAll(() => {
        const app = new App();
        const stack = new OrderProcessingStack(app, 'TestStack');
        template = Template.fromStack(stack);
    });

    test('state machine is created', () => {
        template.hasResourceProperties('AWS::StepFunctions::StateMachine', {
            StateMachineName: 'OrderProcessing'
        });
    });

    test('state machine has correct timeout', () => {
        template.hasResourceProperties('AWS::StepFunctions::StateMachine', {
            // Check that the state machine definition contains the timeout
        });
    });

    test('Lambda functions have correct memory and timeout', () => {
        template.allResourcesProperties('AWS::Lambda::Function', {
            MemorySize: 512,
            Timeout: 30
        });
    });

    test('IAM role grants necessary permissions', () => {
        template.hasResourceProperties('AWS::IAM::Policy', {
            PolicyDocument: {
                Statement: [
                    {
                        Action: ['dynamodb:GetItem', 'dynamodb:PutItem', 'dynamodb:UpdateItem'],
                        Effect: 'Allow'
                    }
                ]
            }
        });
    });
});

Testing Retry and Wait States

Wait states and retry logic are critical but often untested. Step Functions Local lets you test these:

def test_retry_on_transient_failure(sf_client, state_machine_arn):
    """
    Verify that transient Lambda errors trigger retries and eventually succeed.
    Use mock config with a step that fails twice then succeeds.
    """
    response = sf_client.start_execution(
        stateMachineArn=state_machine_arn,
        name='RetryTest',
        input=json.dumps({'orderId': 'order-retry-test'})
    )

    result = wait_for_execution(sf_client, response['executionArn'], timeout=60)

    # With 3 retries configured, 2 failures + 1 success should still SUCCEED
    assert result['status'] == 'SUCCEEDED'

    # Verify retries happened
    history = sf_client.get_execution_history(executionArn=response['executionArn'])
    lambda_failed_events = [
        e for e in history['events']
        if e['type'] == 'LambdaFunctionFailed'
    ]
    # 2 failures before the final success
    assert len(lambda_failed_events) == 2

CI/CD Integration

name: Step Functions Tests

on: [push, pull_request]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - run: pip install -r requirements-dev.txt
      - run: pytest tests/unit/ -v

  state-machine-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Start Step Functions Local
        run: |
          docker run -d -p 8083:8083 \
            -v $(pwd)/tests/mocks:/mocks \
            amazon/aws-stepfunctions-local \
            --mock-config /mocks/MockConfigFile.json
      - name: Wait for Step Functions Local
        run: |
          until curl -s http://localhost:8083; do sleep 1; done
      - run: pip install -r requirements-dev.txt
      - run: pytest tests/state-machine/ -v
        env:
          SF_ENDPOINT: http://localhost:8083
          AWS_DEFAULT_REGION: us-east-1
          AWS_ACCESS_KEY_ID: test
          AWS_SECRET_ACCESS_KEY: test

Continuous Monitoring in Production

Step Functions provides detailed execution history, but it doesn't alert you when a workflow starts failing more than expected. After deploying, use HelpMeTest to run synthetic test executions against your live state machine on a schedule, verifying the full workflow completes successfully. This catches regressions from Lambda deployments or downstream service changes before they accumulate into a user-visible incident.

Common Pitfalls

Input/output path confusion: Step Functions uses InputPath, OutputPath, and ResultPath to shape data flow between steps. These are easy to get wrong and hard to debug. Test each step's data transformation explicitly with a known input and verify the exact output shape that reaches the next step.

Catch vs. Retry ordering: When both Catch and Retry are defined, retries run first. If you expect a Catch to fire on the first failure, you need to exhaust the retries first (or not configure retries for that error type).

Map state concurrency: If your state machine uses a Map state for parallel processing, test with both empty arrays and arrays that exceed your MaxConcurrency setting.

Execution history limits: Step Functions stores up to 25,000 history events per execution. For very long-running or deeply nested state machines, test that you stay within this limit.

Timeouts: State machines and individual task states have independent timeout settings. Test that your individual step timeouts are shorter than the state machine timeout, and that timeout errors are properly caught.

The combination of unit tests for Lambda steps, state machine logic tests with Step Functions Local and mock configs, and integration tests against a real or LocalStack AWS environment gives you confidence that your workflows behave correctly — from the happy path all the way through your error handling and retry logic.

Read more

Start now free