AWS and Moto

Python is an excellent tool to implement applications running on AWS, and also testing them.

Introduction

In my previous blog post Python Experiences, I wrote about some recent experiences using Python in production. In this new blog post I write about my experiences using Python to test an application running on AWS infrastructure.

Moto Testing Library

Python is an excellent dynamic language to implement applications running on AWS infrastructure. There is an excellent library for manipulating AWS resources using python: boto3. For testing application logic, I used the moto testing library. Moto is just great. Using Moto, you can setup a fixture that is a one-to-one copy of your production AWS infrastructure. Therefore you can create various testing scenarios (happy-day scenarios and non-happy day scenarios) that test your application running on AWS infrastructure. I realized that you can actually develop your application logic using the testing fixture, i.e. you kind of have the AWS resources as virtual entities in the RAM of your workstation. True test-driven-development!

Example

Let’s create a very simple example using an AWS S3 bucket. You can find this example in my repo: python/moto.

Setup

Checkout the repository and go to the testing/moto directory. Then create a python virtual environment for yourself, e.g.:

python3.12 -m venv .venv

Activate the python virtual environment and install the requirements:

source .venv/bin/activate
pip install -r tests/requirements.txt

I used the latest package versions when writing this blog post, requirements.txt:

pytest==8.1.1
boto3==1.34.86
moto==5.0.5

Application

Let’s use a very simple application as an example. The application processes objects in a bucket, in a prefix called unprocessed. Once the object is processed, the application moves the object to a processed prefix. Let’s see the Python code in process_object.py:

import boto3
from botocore.exceptions import ClientError

s3_client = boto3.client("s3")


def _check_object(bucket: str, key: str) -> bool:
    """Check if the object exists in the bucket."""
    try:
        s3_client.head_object(Bucket=bucket, Key=key)
        return True
    except ClientError:
        return False


def _some_application_logic_with_the_object(bucket: str, key: str):
    """Process the object in the bucket."""
    # Here we would add the code to do some application logic with the object,
    # in this demo just print a message.
    print(f"Processing object {key} in bucket {bucket}")
    # Simulate that the object processing failed for a specific key.
    if "fail" in key:
        raise ValueError(f"Some error occurred processing the object: {key}")


def _move_object_to_processed(bucket: str, key: str):
    """Move the object from the unprocessed/ prefix to the processed/ prefix."""
    copy_source = {"Bucket": bucket, "Key": key}
    processed_key = key.replace("unprocessed/", "processed/")
    s3_client.copy_object(CopySource=copy_source,
                          Bucket=bucket,
                          Key=processed_key)
    s3_client.delete_object(Bucket=bucket, Key=key)


def process_object(bucket: str, key: str):
    """ Main entry point to the module to process the object in the bucket."""
    if not _check_object(bucket, key):
        raise ValueError(f"Object {key} not found in bucket {bucket}")
    _some_application_logic_with_the_object(bucket, key)
    _move_object_to_processed(bucket, key)

As you can see, we simulate failed processing for all files that have fail in the bucket key (for demonstration purposes).

Testing

Then let’s see the test code in test_process_object.py.

First the test fixture:

@pytest.fixture(name="s3_client", scope="function")
def _s3_client():
    with mock_aws():
        yield boto3.client("s3", region_name="us-east-1")


@pytest.fixture(name="bucket_name", scope="function")
def _process_objects_bucket_name(s3_client):
    bucket_name = "some-bucket"
    s3_client.create_bucket(Bucket=bucket_name)
    return bucket_name

As you can see the test fixture is really simple. With this test fixture when your application uses boto3 api to manipulate AWS resources, we actually use mocked objects with the moto library.

Then let’s create a happy-day scenario:

def test_process_object_happy_day_scenario(s3_client, bucket_name):
    bucket = bucket_name
    test_file_name = "object1.txt"
    unprocessed_key = f"unprocessed/{test_file_name}"
    processed_key = f"processed/{test_file_name}"
    add_object_to_unprocessed_prefix(s3_client, bucket, test_file_name)
    # Verify we have the object in the unprocessed prefix now.
    assert s3_client.head_object(Bucket=bucket, Key=unprocessed_key) is not None
    # Verify the object is not in the processed prefix yet.
    with pytest.raises(s3_client.exceptions.ClientError):
        s3_client.head_object(Bucket=bucket, Key=processed_key)
    # Use our application logic to process the object.
    from bucket.process_object import process_object
    process_object(bucket, unprocessed_key)
    # Verify the object is not in the unprocessed prefix any more.
    with pytest.raises(s3_client.exceptions.ClientError):
        s3_client.head_object(Bucket=bucket, Key=unprocessed_key)
    # Verify we have the object in the processed prefix now.
    assert s3_client.head_object(Bucket=bucket, Key=processed_key) is not None

We add a test file to the bucket, then we do our application logic with the object, and finally test that the object has been moved from the unprocessed prefix to the processed prefix.

Then let’s test a non-happy day scenario in which the processing fails for some reason (with fail1.txt file):

def test_process_object_failed_scenario(s3_client, bucket_name):
    bucket = bucket_name
    test_file_name = "fail1.txt"
    unprocessed_key = f"unprocessed/{test_file_name}"
    processed_key = f"processed/{test_file_name}"
    add_object_to_unprocessed_prefix(s3_client, bucket, test_file_name)
    # Verify we have the object in the unprocessed prefix now.
    assert s3_client.head_object(Bucket=bucket, Key=unprocessed_key) is not None
    # Verify the object is not in the processed prefix yet.
    with pytest.raises(s3_client.exceptions.ClientError):
        s3_client.head_object(Bucket=bucket, Key=processed_key)
    # Use our application logic to process the object.
    from bucket.process_object import process_object
    # Test that the processing fails.
    with pytest.raises(ValueError, match=f"Some error occurred processing the object: {unprocessed_key}"):
        process_object(bucket, unprocessed_key)
    # Verify the object is still in the unprocessed prefix since the processing failed.
    assert s3_client.head_object(Bucket=bucket, Key=unprocessed_key) is not None    
    # Verify the object is not moved into the processed prefix since the processing failed.
    with pytest.raises(s3_client.exceptions.ClientError):
        s3_client.head_object(Bucket=bucket, Key=processed_key)

As you can see we verify that the error is exactly the same error that the application logic raises, and that the object was not moved from the unprocessed prefix to the processed prefix, since the object processing failed.

And finally, another non-happy day scenario in which we cannot find the object in the bucket:

def test_process_object_does_not_exist_scenario(s3_client, bucket_name):
    bucket = bucket_name
    test_file_name = "not-exists1.txt"
    unprocessed_key = f"unprocessed/{test_file_name}"
    # Use our application logic to process the object.
    from bucket.process_object import process_object
    # Test that the processing fails since the object does not exist in the bucket and key.
    with pytest.raises(ValueError, match=f"Object {unprocessed_key} not found in bucket {bucket}"):
        process_object(bucket, unprocessed_key)

Running the Tests

You can run the tests with command:

python -m pytest tests

Conclusions

This was a really simplified example for demonstration purposes. In real life I just implemented a much more complex AWS infrastructure and application in which pictures get identified using the AWS Rekognition service. In this system, clients send pictures to a S3 bucket. The bucket has a trigger that when a new picture is uploaded the bucket creates an event to SQS queue. The application reads the SQS queue, reads the picture from the bucket and sends the picture to the Rekognition service. I managed to create a test fixture for all these AWS entities: the S3 bucket, the triggering mechanism, the SQS queue and the Rekognition service. I have extensive tests for all happy day and non-happy day scenarios and modifying the application code is safe since I have a good code coverage using my tests.

The writer is working at a major international IT corporation building cloud infrastructures and implementing applications on top of those infrastructures.

Kari Marttila

Kari Marttila’s Home Page in LinkedIn: https://www.linkedin.com/in/karimarttila/