This post shows you a short example of how to use the Python module moto to mock a SQS queue. This technique is useful when writing code level tests for applications hosted on an AWS stack. We will work with a method that takes a string as an input, processes the string and then writes it to an SQS queue. We will use moto to mock the SQS queue and verify that the right message gets sent to the queue.

Background

You can safely skip this section if you do not care for where this work came from. Qxf2 failed on a client project recently. We had test an application that was hosted entirely on the AWS stack. The application was made of several pipelines that each consisted of loosely coupled microservices (lambdas), used s3 buckets as a datastore and were connected by messaging queues (SQS) and message broadcasters (SNS). While this sort of arrangement sounds straightforward to test, we struggled massively. We quickly realized that we lacked the right tooling to aid our testing. So, for the last few months, we have been working on fixing that hole. This post documents one of our learnings along the way. We hope that our learning helps other professional testers who are also struggling to test similar applications.

The method under test

Our method under test is a simplified version of what writing to a SQS queue using Python looks like. We take a couple of inputs, create a message out of it and then write it to a queue. While this example looks contrived, it covers the most critical case I want to show as part of this blog post.

QUEUE_URL = 'blah blah blah'
def write_message(daily_message, channel):
    "Send a message to Skype Sender"
    sqs = boto3.client('sqs')
    message = str({'msg':f'{daily_message}', 'channel':channel})
    sqs.send_message(QueueUrl=QUEUE_URL, MessageBody=(message))

Let us assume this code exists in a file called daily_messages.py.

Using moto’s @mock_sqs

The Python module moto makes it really easy for us to mock this operation. moto provides the @mock_sqs decorator that mocks out an SQS queue. Our way of writing this one test will involve the following steps:

1. Decorate the test method with @mock_sqs
2. Create an SQS resource
3. Create a queue
4. Force daily_messages.write_message() to use the newly created queue
5. Create inputs for daily_messages.write_message()
6. Arrive at an expected result
7. Call daily_messages.write_message()
8. Read the queue for messages
9. Verify that the message body matches the expected result
 

1. Decorate the test method with @mock_sqs

Create a new file called test_daily_messages.py and add the following lines:

import boto3
from moto import mock_sqs
import daily_messages
 
@mock_sqs
def test_write_message_valid():
    "Test the write_message method with a valid message"

At this stage, we have imported the necessary modules and decorated our test method with @mock_sqs.

2. Create an SQS resource

Now, within the test_write_message_valid() method, create an SQS resource like this

sqs = boto3.resource('sqs')

3. Create a queue

You can create a queue using the SQS resource like this

queue = sqs.create_queue(QueueName='test-skype-sender')

4. Force daily_messages.write_message() to use the newly created queue

If you notice daily_messages.py uses a global variable called QUEUE_URL to select the queue being used. So let us obtain the queue URL of our newly created queue and set that as daily_messages.QUEUE_URL.

daily_messages.QUEUE_URL = queue.url

Tip: To find out more about what the queue object contains, you can print out queue.__dict__ and queue.attributes.

5. Create inputs for daily_messages.write_message()

This is the part where testers shine. For this example, I will just show one tame combination of input to daily_messages.write_message() but when testing a real app, go crazy with your inputs!

skype_message = 'Testing with a valid message'
channel = 'test'

6. Arrive at an expected result

Understand the transformation that happens in daily_messages.write_message() to arrive at an expected result. We notice daily_messages.write_message() takes the two input arguments, creates a simple dictionary with keys msg and channel and finally converts the dictionary into a string. To arrive at our expected result, we will simply do the same.

expected_message = str({'msg':f'{skype_message}', 'channel':channel})

For folks used to the Arrange, Act, Assert style, we have just completed the Arrange portion.

7. Call daily_messages.write_message()

daily_messages.write_message(skype_message, channel)

And at this point, when daily_messages.write_message() is executed, it should end up populating the SQS queue we created in step 3. We are done with the Act portion of the Arrange, Act, Assert pattern.

8. Read the queue for messages

moto follows the same patterns as boto3. So if you are familiar with boto3, you could have simply guessed that queue.receive_messages() was possible on the queue object.

sqs_messages = queue.receive_messages()

9. Verify that the message body matches the expected result

Let’s assert that the body of the message looks right

assert sqs_messages[0].body == expected_message, 'Message in skype-sender does not match expected'

Tip: You can also assert other things. For example, you can verify that exactly one message was sent using assert len(sqs_messages) == 1, 'Expected exactly one message in SQS'.

Putting it all together

Here is how our test looks:

"""
Example of using moto to mock out an SQS queue
"""
 
import boto3
from moto import mock_sqs
import daily_messages
 
@mock_sqs
def test_write_message_valid():
    "Test the write_message method with a valid message"
    sqs = boto3.resource('sqs')
    queue = sqs.create_queue(QueueName='test-skype-sender')
    daily_messages.QUEUE_URL = queue.url
    skype_message = 'Testing with a valid message'
    channel = 'test'
    expected_message = str({'msg':f'{skype_message}', 'channel':channel})
    daily_messages.write_message(skype_message, channel)
    sqs_messages = queue.receive_messages()
    assert sqs_messages[0].body == expected_message, 'Message in skype-sender does not match expected'
    print(f'The message in skype-sender SQS matches what we sent')
    assert len(sqs_messages) == 1, 'Expected exactly one message in SQS'
    print(f'\nExactly one message in skype-sender SQS')

If you are a tester working on testing different aspects of an AWS pipeline, consider using mocking as part of your tests. I hope this article simplifies the thought process behind using moto for mocking an AWS SQS queue.

Related posts:

How to use moto with AWS DynamoDB Filter Messages from the AWS SQS Queue using Python and asyncio Holiday Reminder Skype Bot