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.
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 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
- 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.
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_credentialsfixture 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.
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_testthat 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.
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 ====================================
Similarly, let’s create a client to handle SQS functionality. Let’s say we have a file called
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"]["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_testthat 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 ====================================
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.