Testing AWS Step Functions

Python unit tests

1. Install python-on-whales

poetry add --group test python-on-whales

2. Create a session scope fixture that will spin up the docker container with local implementation of Step Functions (amazon/aws-stepfunctions-local). Mock config file can be used to provide mocks for any of the step function state.

The tree of my Terraform module:

module_name
├── envs
│   ├── dev.tfvars
│   ├── production.tfvars
│   └── staging.tfvars
├── main_step_function.tf
├── main.tf
├── outputs.tf
├── step_function_definition.asl.json
├── tests
│   ├── __init__.py
│   ├── mocks.json
│   └── test_step_function.py
└── variables.tf

# tests/test_step_function.py
from pathlib import Path
from collections.abc import Generator
 
import pytest
from python_on_whales import docker
 
LOCAL_SFN_PORT = 8083
MOCK_CONFIG_PATH = Path(__file__).parent / "mocks.json"
 
@pytest.fixture(scope="session", autouse=True)
def _sfn_local_container() -> Generator[None, None, None]:
    docker.pull("amazon/aws-stepfunctions-local")
 
    # Run the step function local container
    mock_config_container_path = Path("/home/StepFunctionsLocal/MockConfigFile.json")
    with docker.run(
        "amazon/aws-stepfunctions-local",
        mounts=[
            [
                "type=bind",
                "readonly",
                f"source={str(MOCK_CONFIG_PATH)}",
                f"destination={str(mock_config_container_path)}",
            ],
        ],
        envs={
            "SFN_MOCK_CONFIG": str(mock_config_container_path),
        },
        publish=[(LOCAL_SFN_PORT, LOCAL_SFN_PORT)],
        detach=True,
    ) as _:  # type: ignore
        # yield fixture and then remove the container
        yield

3. Create a fixture that will initialize boto3 stepfunctions client targeting local endpoint (with a port we specified previously):

# tests/test_step_function.py
import boto3
 
from mypy_boto3_stepfunctions import SFNClient
 
LOCAL_SFN_ENDPOINT = f"http://localhost:{LOCAL_SFN_PORT}"
 
@pytest.fixture(scope="module")
def sfn_client() -> SFNClient:
    return boto3.client("stepfunctions", endpoint_url=LOCAL_SFN_ENDPOINT)

4. Then use this fixture in the test to start the execution of a step function:

# tests/test_step_function.py
def test_sfn_can_be_executed(sfn_client: SFNClient, sfn_arn: str) -> None:
    # GIVEN
    # WHEN the step function is executed
    response = sfn_client.start_execution(stateMachineArn=f"{sfn_arn}")
 
    # THEN a valid ARN of the execution is returned
    assert response["executionArn"].startswith(
        "arn:aws:states:us-east-1:123456789012:execution:",
    )

To start the execution using mocked test case, append test case name after the arn of the step function:

# tests/test_step_function.py
def test_sfn_happy_path_finishes_successfully(
    sfn_client: SFNClient,
    sfn_arn: str,
) -> None:
    # GIVEN
    mock_path = "HappyPath"
 
    # WHEN the step function is executed
    response = sfn_client.start_execution(stateMachineArn=f"{sfn_arn}#{mock_path}")
 
    # THEN the status of the execution is successful
    execution = describe_execution(sfn_client, response["executionArn"])
 
    assert execution
    assert execution["status"] == "SUCCEEDED"

Manual

1. Setup up with Docker

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

Task:

  run-step-function-local:
    desc: Run step function locally in the Docker container.
    cmds:
      - docker pull amazon/aws-stepfunctions-local
      - docker run -p 8083:8083 amazon/aws-stepfunctions-local
 

…

Links

• AWS Blog - Mocking service integrations with AWS Step Functions Local
• AWS Step Functions Docs - Testing state machines locally
• Using Mocked Service Integrations
• https://notes.paulswail.com/public/Testing+Step+Functions
• Python on Whales docs