Introduction

AWS’ Boto library is used commonly to integrate Python applications with various AWS services such as EC2, S3, and SQS amongst others. I have generally avoided writing unit-tests for application code that interacts with the boto library because of the complexity involved in mocking and testing these functions.

However, I recently tried out the Moto library which makes it easy to mock AWS services and test code that interacts with AWS.

Some of the benefits of using Moto:

  • Testing code that interacts with AWS. Instead of having to test your code in an AWS environment, test AWS interactions locally.
  • Easy to learn and get started with.
  • Extensive coverage of AWS services.

In this article, we will look at how to add unit tests using Moto.

Prerequisites

We need to install the following libraries to run the application:

  • boto3: AWS SDK for Python
  • moto: Mock AWS Services
  • Pytest: Test framework for Python

These libraries can be installed via pip:

pip install boto3 moto pytest

Moto

Moto is a python library that makes it easy to mock AWS services. Moto works well because it mocks out all calls to AWS automatically without requiring any dependency injection.

There are a couple of things that need to be kept in mind while using moto:

  • Use decorators: Moto provides decorators for mocking out specific services. Test functions need to be annotated with the appropriate decorator. For E.g. if we want to test S3 functionality, we would use mock_s3.
  • Setup dummy credentials: It is recommended to set up dummy environment variables for AWS access to ensure we don’t accidentally mutate any production resources. Any AWS connections using boto3 should be set up after the mocking has been setup.

Moto works well with Pytest fixtures and makes testing AWS functionality pretty straightforward.

Getting Started

Conftest file

Pytest makes it easy to share fixtures across various modules if we add them to a file named conftest.py. Let’s create such a file with the following content:


import boto3
import os
import pytest

from moto import mock_s3, mock_sqs


@pytest.fixture
def aws_credentials():
    """Mocked AWS Credentials for moto."""
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_SECURITY_TOKEN"] = "testing"
    os.environ["AWS_SESSION_TOKEN"] = "testing"


@pytest.fixture
def s3_client(aws_credentials):
    with mock_s3():
        conn = boto3.client("s3", region_name="us-east-1")
        yield conn


@pytest.fixture
def sqs_client(aws_credentials):
    with mock_sqs():
        conn = boto3.client("sqs", region_name="us-east-1")
        yield conn

Things to note:

  • We are mocking AWS credentials in the aws_credentials fixture which is being used by the various client fixtures.
  • Client fixtures for S3 and SQS which use the mock credentials. These client fixtures will be used later to test our application code.

S3

Let’s create a file called s3.py with the following content:

import boto3


class MyS3Client:
    def __init__(self, region_name="us-east-1"):
        self.client = boto3.client("s3", region_name=region_name)

    def list_buckets(self):
        """Returns a list of bucket names."""
        response = self.client.list_buckets()
        return [bucket["Name"] for bucket in response["Buckets"]]

    def list_objects(self, bucket_name, prefix):
        """Returns a list all objects with specified prefix."""
        response = self.client.list_objects(
            Bucket=bucket_name,
            Prefix=prefix,
        )
        return [object["Key"] for object in response["Contents"]]

We have created an S3 client which performs two key functions:

  • Returns the list of buckets in our AWS account
  • Returns the list of objects that match a particular prefix

Now, to test our S3 client, let’s create a file called test_s3.py with the following content:

import pytest
from tempfile import NamedTemporaryFile

from s3 import MyS3Client


@pytest.fixture
def bucket_name():
    return "my-test-bucket"


@pytest.fixture
def s3_test(s3_client, bucket_name):
    s3_client.create_bucket(Bucket=bucket_name)
    yield


def test_list_buckets(s3_client, s3_test):
    my_client = MyS3Client()
    buckets = my_client.list_buckets()
    assert buckets == ["my-test-bucket"]


def test_list_objects(s3_client, s3_test):
    file_text = "test"
    with NamedTemporaryFile(delete=True, suffix=".txt") as tmp:
        with open(tmp.name, "w", encoding="UTF-8") as f:
            f.write(file_text)

        s3_client.upload_file(tmp.name, "my-test-bucket", "file12")
        s3_client.upload_file(tmp.name, "my-test-bucket", "file22")

    my_client = MyS3Client()
    objects = my_client.list_objects(bucket_name="my-test-bucket", prefix="file1")
    assert objects == ["file12"]

Things to note:

  • s3_test_: Before we can test the functionality in our application code, we need to create a mock S3 bucket. We have set up a fixture called s3_test that will first create a bucket.
  • test_list_buckets: In this test, we assert that the list of buckets our client retrieved is what we expect.
  • test_list_objects: In this test, we created two temporary files with different keys and then tested that we retrieved the file with the matching prefix.

Testing

We can run the unit tests by running the following command:

pytest test_s3.py

======================================== test session starts =========================================
platform linux -- Python 3.7.3, pytest-6.1.2, py-1.9.0, pluggy-0.13.1
rootdir: /home/abhishek/Projects/aws-projects/aws-boto-unit-tests
collected 2 items

test_s3.py ..                                                                                  [100%]

========================================== warnings summary ==========================================
test_s3.py::test_list_buckets
  /home/abhishek/.virtualenvs/aws-projects/lib/python3.7/site-packages/boto/plugin.py:40: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses
    import imp

-- Docs: https://docs.pytest.org/en/stable/warnings.html
==================================== 2 passed, 1 warning in 0.40s ====================================

SQS

Similarly, let’s create a client to handle SQS functionality. Let’s say we have a file called sqs.py

import boto3


class MySQSClient:
    def __init__(self, region_name="us-east-1"):
        self.client = boto3.client("sqs", region_name=region_name)

    def get_queue_url(self, queue_name):
        response = self.client.create_queue(
            QueueName=queue_name
        )
        return response["QueueUrl"]

    def receive_message(self, queue_url):
        return self.client.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=1,
        )

We have created an SQS client which performs two key functions:

  • Returns the URL of the queue
  • Retrieves a message from the queue

Now, to test our SQS client, let’s create a file called test_sqs.py with the following content:


import pytest

from sqs import MySQSClient


@pytest.fixture
def queue_name():
    return "my-test-queue"


@pytest.fixture
def sqs_test(sqs_client, queue_name):
    sqs_client.create_queue(QueueName=queue_name)
    yield


def test_get_queue_url(sqs_client, sqs_test):
    sqs_client = MySQSClient()
    queue_url = sqs_client.get_queue_url(queue_name="my-test-queue")
    assert "my-test-queue" in queue_url


def test_receive_message(sqs_client, sqs_test):
    client = MySQSClient()
    queue_url = client.get_queue_url(queue_name="blah")

    sqs_client.send_message(
        QueueUrl=queue_url,
        MessageBody="derp"
    )

    response = client.receive_message(queue_url=queue_url)
    assert response["Messages"][0]["Body"] == "derp"

Things to note:

  • sqs_test_: Before we can test the functionality in our application code, we need to create a mock SQS queue. We have set up a fixture called sqs_test that will first create the queue.
  • test_get_queue_url: In this test, we assert that the URL of the queue contains the name of the queue we created.
  • test_receive_message: In this test, we first enqueue a message into the queue using the boto3 client directly. Then, we use our SQS client to retrieve the message and assert that the message equals what we had initially enqueued.

Running the test

We can run the test by running the following command:

pytest test_sqs.py
======================================== test session starts =========================================
platform linux -- Python 3.7.3, pytest-6.1.2, py-1.9.0, pluggy-0.13.1
rootdir: /home/abhishek/Projects/aws-projects/aws-boto-unit-tests
collected 2 items

test_sqs.py ..                                                                                 [100%]

========================================== warnings summary ==========================================
test_sqs.py::test_get_queue_url
  /home/abhishek/.virtualenvs/aws-projects/lib/python3.7/site-packages/boto/plugin.py:40: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses
    import imp

-- Docs: https://docs.pytest.org/en/stable/warnings.html
==================================== 2 passed, 1 warning in 0.67s ====================================

Conclusion

Moto and Pytest make it pretty easy to unit test AWS-specific application code. It has already helped me become more confident about code that is being deployed to production and makes it possible to test such locally instead of in an AWS environment.