Unit Testing AWS – Moto

Shivam Bhatnagar

The moto library provides a simple wrapper for mocking AWS resources so that we are not affecting live AWS resources in our unit tests

Unit tests check the behaviour of a function/method based on a given input, these tests should always run in isolation and not be dependent on external resources (e.g. databases, AWS resources, APIs).

Traditionally external systems can be mocked so that tests do not make live calls to live systems (e.g. using responses to mock APIs). Mocking provides a safe framework that isolates our tests away from live systems and lets us specify attributes of a given system e.g. return values.

The moto library provides a simple wrapper for mocking AWS resources so that we are not affecting live AWS resources in our unit tests

Installing Moto

Moto can be installed using pip, or by adding moto to your project requirements

Installing moto for all resources

pip install 'moto[all]'

Installing moto for specific resources – if you need to minimise project dependencies

pip install 'moto[ec2, s3]'

Writing a test using Moto

The code can be found here

Given the following function – which moves a file from one S3 bucket to another

import boto3
import os

def move_1_file_to_new_bucket(source_s3_bucket: str, source_s3_file, destination_s3_bucket: str,
                             destination_folder: str = '', destination_s3_file=None):
    """
    Move 1 file from one s3 bucket to another
    :param: source_s3_bucket: source bucket name
    :param: source_s3_file: source file path
    :param: destination_s3_bucket: destination bucket name
    :param: destination_folder
    :param: destination_s3_file: optional new file name
    :return: list of destination file paths
    """
    s3_wranglers_dev = boto3.resource('s3')
    if destination_s3_file is not None:
        destination_key = destination_folder + source_s3_file
    try:
        copy_source = {
            'Bucket': source_s3_bucket,
            'Key': source_s3_file
        }
        encrypt_files = {
            'ServerSideEncryption': 'aws:kms',
            'SSEKMSKeyId': os.getenv('SSEKMSKeyId')
        }
        s3_wranglers_dev.meta.client.copy(copy_source, destination_s3_bucket, destination_key,
                                            ExtraArgs=encrypt_files)
    except Exception:
        print(f'Error moving {source_s3_bucket}/{source_s3_file} to {destination_s3_bucket}/{destination_key}')
        raise
    print(f'{source_s3_bucket}/{source_s3_file} copied to location: '
            f'{destination_s3_bucket}/{destination_key}')
    return destination_key

We can write a unit-test as follows

import pytest
import move_1_file_to_new_bucket
import boto3
from moto import mock_s3

class TestMoveFile:

    @mock_s3
    def test_move_1_file_to_new_bucket(self, monkeypatch):

        # Arrange
        s3_resource = boto3.resource("s3")
        s3_resource.create_bucket(Bucket='source_bucket')
        s3_resource.create_bucket(Bucket='target_bucket')


        file_data = b'This is my source file data'
        file_obj = s3_resource.Object('source_bucket', 'source_file')
        file_obj.put(Body=file_data)

        monkeypatch.setenv('SSEKMSKeyId', 'ssekmskey')

        # Act
        move_1_file_to_new_bucket.move_1_file_to_new_bucket(
            source_s3_bucket='source_bucket',
            source_s3_file='source_file',
            destination_s3_bucket='target_bucket',
            destination_s3_file='target_file'
        )

        # List contents of target bucket (to check if file was uploaded)
        target_bucket = s3_resource.Bucket('target_bucket')
        files_in_target = []
        for my_bucket_object in target_bucket.objects.all():
            files_in_target.append(my_bucket_object.key)
            
        # Assert
        assert 'source_file' in files_in_target