AWS

Testing AWS Step Functions: Local Testing with Step Functions Local and Moto

HelpMeTest

16 May 2026 — 4 min read

AWS Step Functions orchestrate Lambda functions, ECS tasks, and other AWS services into durable workflows. Testing them requires either expensive real AWS calls or local simulation. AWS provides Step Functions Local for exactly this purpose, and moto mocks the underlying AWS services.

Testing Approaches

Step Functions testing has three levels:

Definition validation — catch typos and invalid ASL (Amazon States Language) before deployment
Local simulation — run state machines locally with Step Functions Local
Integration testing — execute against real (or Moto-mocked) AWS services

Validating State Machine Definitions

Use the AWS CLI or a JSON schema validator to catch definition errors early:

# Validate with AWS CLI (calls the validation API, no execution)
aws stepfunctions create-state-machine \
  --name validate-only \
  --definition file://state-machine.json \
  --role-arn arn:aws:iam::123456789012:role/test-role \
  --no-cli-pager 2>&1 <span class="hljs-pipe">| grep -i error

Or validate locally using asl-validator:

npm install -g asl-validator
asl-validator --json-definition state-machine.json

Example state machine for testing:

{
  "Comment": "Order processing workflow",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:::function:validate-order",
      "Next": "ProcessPayment",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["ValidationError"],
          "Next": "OrderFailed"
        }
      ]
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:::function:process-payment",
      "End": true
    },
    "OrderFailed": {
      "Type": "Fail",
      "Error": "OrderValidationFailed",
      "Cause": "Order did not pass validation"
    }
  }
}

Step Functions Local

Step Functions Local is a Java application that simulates the Step Functions API locally:

# Download and run
docker pull amazon/aws-stepfunctions-local
docker run -p 8083:8083 amazon/aws-stepfunctions-local

Or with a config file:

# sfn-local.conf
LambdaEndpoint=http://host.docker.internal:3001
SFNMockConfigFile=/home/StepFunctionsLocal/MockConfigFile.json

docker run -p 8083:8083 \
  -e AWS_DEFAULT_REGION=us-east-1 \
  -v $(pwd)/sfn-local.conf:/home/StepFunctionsLocal/aws-stepfunctions-local-credentials.txt \
  amazon/aws-stepfunctions-local

Mock Config for Lambda Calls

Step Functions Local can use mock Lambda responses instead of calling real functions:

{
  "StateMachines": {
    "OrderWorkflow": {
      "TestCases": {
        "HappyPath": {
          "ValidateOrder": "ValidationSuccess",
          "ProcessPayment": "PaymentSuccess"
        },
        "ValidationFailure": {
          "ValidateOrder": "ValidationFailure"
        }
      }
    }
  },
  "MockedResponses": {
    "ValidationSuccess": {
      "Return": {
        "StatusCode": 200,
        "Payload": "{\"valid\": true, \"orderId\": \"order-123\"}"
      }
    },
    "ValidationFailure": {
      "Throw": {
        "Error": "ValidationError",
        "Cause": "Missing required field: quantity"
      }
    },
    "PaymentSuccess": {
      "Return": {
        "StatusCode": 200,
        "Payload": "{\"transactionId\": \"txn-456\", \"status\": \"charged\"}"
      }
    }
  }
}

Running Tests Against Step Functions Local

import boto3
import json
import time
import pytest

@pytest.fixture
def sfn_client():
    return boto3.client(
        "stepfunctions",
        endpoint_url="http://localhost:8083",
        region_name="us-east-1",
        aws_access_key_id="local",
        aws_secret_access_key="local",
    )

@pytest.fixture
def state_machine_arn(sfn_client):
    with open("state-machine.json") as f:
        definition = f.read()

    response = sfn_client.create_state_machine(
        name="OrderWorkflow",
        definition=definition,
        roleArn="arn:aws:iam::123456789012:role/test-role",
        type="STANDARD",
    )
    arn = response["stateMachineArn"]
    yield arn

    # Cleanup
    sfn_client.delete_state_machine(stateMachineArn=arn)

def test_order_workflow_happy_path(sfn_client, state_machine_arn):
    response = sfn_client.start_execution(
        stateMachineArn=state_machine_arn,
        name="test-happy-path",
        input=json.dumps({"orderId": "order-123", "amount": 99.99}),
        traceHeader="test=HappyPath",  # selects mock config
    )
    execution_arn = response["executionArn"]

    # Poll for completion
    for _ in range(30):
        status = sfn_client.describe_execution(executionArn=execution_arn)
        if status["status"] in ("SUCCEEDED", "FAILED", "TIMED_OUT", "ABORTED"):
            break
        time.sleep(0.5)

    assert status["status"] == "SUCCEEDED"
    output = json.loads(status["output"])
    assert output["transactionId"] == "txn-456"

def test_order_workflow_validation_failure(sfn_client, state_machine_arn):
    response = sfn_client.start_execution(
        stateMachineArn=state_machine_arn,
        name="test-validation-failure",
        input=json.dumps({"orderId": "bad-order"}),
        traceHeader="test=ValidationFailure",
    )
    execution_arn = response["executionArn"]

    for _ in range(30):
        status = sfn_client.describe_execution(executionArn=execution_arn)
        if status["status"] in ("SUCCEEDED", "FAILED", "TIMED_OUT", "ABORTED"):
            break
        time.sleep(0.5)

    assert status["status"] == "FAILED"
    assert "OrderValidationFailed" in status.get("error", "")

Testing with Moto

moto is a Python library that mocks AWS services in-process. It's faster than Step Functions Local and requires no Docker:

pip install moto[stepfunctions,lambda]

import json
import boto3
import pytest
from moto import mock_aws

@mock_aws
def test_state_machine_creation():
    client = boto3.client("stepfunctions", region_name="us-east-1")

    response = client.create_state_machine(
        name="TestWorkflow",
        definition=json.dumps({
            "StartAt": "HelloWorld",
            "States": {
                "HelloWorld": {
                    "Type": "Pass",
                    "Result": "Hello, World!",
                    "End": True
                }
            }
        }),
        roleArn="arn:aws:iam::123456789012:role/test-role",
    )

    assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
    assert "stateMachineArn" in response

@mock_aws
def test_execution_lifecycle():
    client = boto3.client("stepfunctions", region_name="us-east-1")

    sm = client.create_state_machine(
        name="SimpleWorkflow",
        definition=json.dumps({
            "StartAt": "Process",
            "States": {
                "Process": {"Type": "Pass", "End": True}
            }
        }),
        roleArn="arn:aws:iam::123456789012:role/test-role",
    )

    exec_response = client.start_execution(
        stateMachineArn=sm["stateMachineArn"],
        input=json.dumps({"key": "value"}),
    )

    exec_arn = exec_response["executionArn"]
    history = client.get_execution_history(executionArn=exec_arn)

    event_types = [e["type"] for e in history["events"]]
    assert "ExecutionStarted" in event_types

Unit Testing State Machine Logic

For complex Choice states and error handling, test the logic by examining execution history:

@mock_aws
def test_choice_state_routing():
    client = boto3.client("stepfunctions", region_name="us-east-1")

    # State machine with choice routing
    definition = {
        "StartAt": "CheckAmount",
        "States": {
            "CheckAmount": {
                "Type": "Choice",
                "Choices": [
                    {
                        "Variable": "$.amount",
                        "NumericGreaterThan": 1000,
                        "Next": "HighValueOrder"
                    }
                ],
                "Default": "StandardOrder"
            },
            "HighValueOrder": {"Type": "Pass", "Result": "high", "End": True},
            "StandardOrder": {"Type": "Pass", "Result": "standard", "End": True},
        }
    }

    sm = client.create_state_machine(
        name="ChoiceTest",
        definition=json.dumps(definition),
        roleArn="arn:aws:iam::123456789012:role/test-role",
    )

    # Test high-value path
    execution = client.start_execution(
        stateMachineArn=sm["stateMachineArn"],
        input=json.dumps({"amount": 1500}),
    )

    description = client.describe_execution(
        executionArn=execution["executionArn"]
    )
    output = json.loads(description["output"])
    assert output == "high"

    # Test standard path
    execution2 = client.start_execution(
        stateMachineArn=sm["stateMachineArn"],
        input=json.dumps({"amount": 500}),
    )

    description2 = client.describe_execution(
        executionArn=execution2["executionArn"]
    )
    output2 = json.loads(description2["output"])
    assert output2 == "standard"

CDK/CloudFormation Testing

If you define Step Functions with CDK, snapshot test the generated CloudFormation:

import pytest
from aws_cdk import App
from aws_cdk.assertions import Template
from myapp.stacks import OrderProcessingStack

def test_state_machine_in_template():
    app = App()
    stack = OrderProcessingStack(app, "OrderProcessingStack")
    template = Template.from_stack(stack)

    # Verify state machine exists
    template.resource_count_is("AWS::StepFunctions::StateMachine", 1)

    # Verify retry configuration
    template.has_resource_properties("AWS::StepFunctions::StateMachine", {
        "DefinitionString": {
            "Fn::Join": ["", [
                # Just check the state machine name appears
                {"Fn::Sub": "..."}
            ]]
        }
    })

CI Integration

# .github/workflows/test.yml
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      stepfunctions-local:
        image: amazon/aws-stepfunctions-local
        ports:
          - 8083:8083

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - run: pip install pytest moto[stepfunctions] boto3

      - name: Wait for Step Functions Local
        run: |
          timeout 30 bash -c 'until curl -s http://localhost:8083; do sleep 1; done'

      - run: pytest tests/stepfunctions/ -v

End-to-End Testing Step Functions Workflows

Local tests validate state machine logic. End-to-end tests verify the full system — the API trigger, Step Functions execution, Lambda behavior, and downstream state changes. HelpMeTest can verify the user-visible outcomes of workflow executions:

Scenario: order processing workflow completes
  Given a customer places an order
  When Step Functions processes the order workflow
  Then the order status updates to "fulfilled" within 2 minutes
  And the customer receives a shipment notification

Key Takeaways

Use asl-validator or the AWS CLI create-state-machine with dry-run to catch definition errors early
Step Functions Local with mock config lets you test retry, error, and catch paths without real Lambda functions
moto is faster for unit testing choice routing, Pass states, and execution history — no Docker required
Use the traceHeader parameter in Step Functions Local to select which mock test case to run
Test CDK-defined state machines with aws_cdk.assertions.Template to catch configuration regressions