{"id":13707,"date":"2020-09-21T08:04:08","date_gmt":"2020-09-21T12:04:08","guid":{"rendered":"https:\/\/qxf2.com\/blog\/?p=13707"},"modified":"2020-09-21T08:04:08","modified_gmt":"2020-09-21T12:04:08","slug":"mocking-out-aws-sqs-using-moto","status":"publish","type":"post","link":"https:\/\/qxf2.com\/blog\/mocking-out-aws-sqs-using-moto\/","title":{"rendered":"Mocking an AWS SQS queue using moto"},"content":{"rendered":"

This post shows you a short example of how to use the Python module moto<\/a> to mock a SQS<\/a> 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.<\/p>\n

\n

Background<\/h3>\n

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. <\/p>\n

\n

The method under test<\/h3>\n

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.<\/p>\n

\r\nQUEUE_URL = 'blah blah blah'\r\ndef write_message(daily_message, channel):\r\n    \"Send a message to Skype Sender\"\r\n    sqs = boto3.client('sqs')\r\n    message = str({'msg':f'{daily_message}', 'channel':channel})\r\n    sqs.send_message(QueueUrl=QUEUE_URL, MessageBody=(message))\r\n<\/pre>\n
Let us assume this code exists in a file called daily_messages.py<\/code>. <\/p>\n
\n
Using moto’s @mock_sqs<\/h3>\n
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:<\/p>\n
1. Decorate the test method with @mock_sqs
\n2. Create an SQS resource
\n3. Create a queue
\n4. Force daily_messages.write_message() to use the newly created queue
\n5. Create inputs for daily_messages.write_message()
\n6. Arrive at an expected result
\n7. Call daily_messages.write_message()
\n8. Read the queue for messages
\n9. Verify that the message body matches the expected result
\n <\/p>\n
1. Decorate the test method with @mock_sqs<\/h5>\n
Create a new file called test_daily_messages.py<\/code> and add the following lines:<\/p>\n
\r\nimport boto3\r\nfrom moto import mock_sqs\r\nimport daily_messages\r\n\r\n@mock_sqs\r\ndef test_write_message_valid():\r\n    \"Test the write_message method with a valid message\"\r\n<\/pre>\n
At this stage, we have imported the necessary modules and decorated our test method with @mock_sqs.<\/p>\n
2. Create an SQS resource<\/h5>\n
Now, within the test_write_message_valid()<\/code> method, create an SQS resource like this<\/p>\n
\r\nsqs = boto3.resource('sqs')\r\n<\/pre>\n
3. Create a queue<\/h5>\n
You can create a queue using the SQS resource like this<\/p>\n
\r\nqueue = sqs.create_queue(QueueName='test-skype-sender')\r\n<\/pre>\n
4. Force daily_messages.write_message() to use the newly created queue<\/h5>\n
If you notice daily_messages.py<\/code> 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<\/code>.<\/p>\n
\r\ndaily_messages.QUEUE_URL = queue.url\r\n<\/pre>\n
Tip: To find out more about what the queue<\/code> object contains, you can print out queue.__dict__<\/code> and queue.attributes<\/code>.<\/p>\n
5. Create inputs for daily_messages.write_message()<\/h5>\n
This is the part where testers shine. For this example, I will just show one tame combination of input to daily_messages.write_message()<\/code> but when testing a real app, go crazy with your inputs!<\/p>\n
\r\nskype_message = 'Testing with a valid message'\r\nchannel = 'test'\r\n<\/pre>\n
6. Arrive at an expected result<\/h5>\n
Understand the transformation that happens in daily_messages.write_message()<\/code> to arrive at an expected result. We notice daily_messages.write_message()<\/code> takes the two input arguments, creates a simple dictionary with keys msg<\/code> and channel<\/code> and finally converts the dictionary into a string. To arrive at our expected result, we will simply do the same.<\/p>\n
\r\nexpected_message = str({'msg':f'{skype_message}', 'channel':channel})\r\n<\/pre>\n
For folks used to the Arrange, Act, Assert<\/b> style, we have just completed the Arrange<\/b> portion.<\/p>\n
7. Call daily_messages.write_message()<\/h5>\n
\r\ndaily_messages.write_message(skype_message, channel)\r\n<\/pre>\n
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<\/b> portion of the Arrange, Act, Assert<\/b> pattern.<\/p>\n
8. Read the queue for messages<\/h5>\n
moto follows the same patterns as boto3. So if you are familiar with boto3, you could have simply guessed that queue.receive_messages()<\/code> was possible on the queue<\/code> object.<\/p>\n
\r\nsqs_messages = queue.receive_messages()\r\n<\/pre>\n
9. Verify that the message body matches the expected result<\/h5>\n
Let’s assert that the body of the message looks right<\/p>\n
\r\nassert sqs_messages[0].body == expected_message, 'Message in skype-sender does not match expected'\r\n<\/pre>\n
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'<\/code>. <\/p>\n
\n
Putting it all together<\/h3>\n
Here is how our test looks:<\/p>\n
\r\n\"\"\"\r\nExample of using moto to mock out an SQS queue\r\n\"\"\"\r\n\r\nimport boto3\r\nfrom moto import mock_sqs\r\nimport daily_messages\r\n\r\n@mock_sqs\r\ndef test_write_message_valid():\r\n    \"Test the write_message method with a valid message\"\r\n    sqs = boto3.resource('sqs')\r\n    queue = sqs.create_queue(QueueName='test-skype-sender')\r\n    daily_messages.QUEUE_URL = queue.url\r\n    skype_message = 'Testing with a valid message'\r\n    channel = 'test'\r\n    expected_message = str({'msg':f'{skype_message}', 'channel':channel})\r\n    daily_messages.write_message(skype_message, channel)\r\n    sqs_messages = queue.receive_messages()\r\n    assert sqs_messages[0].body == expected_message, 'Message in skype-sender does not match expected'\r\n    print(f'The message in skype-sender SQS matches what we sent')\r\n    assert len(sqs_messages) == 1, 'Expected exactly one message in SQS'\r\n    print(f'\\nExactly one message in skype-sender SQS')\r\n<\/pre>\n
\n
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.<\/p>\n
\n","protected":false},"excerpt":{"rendered":"
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. […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[177,240,64,249],"tags":[],"class_list":["post-13707","post","type-post","status-publish","format-standard","hentry","category-aws","category-aws-sqs","category-mock","category-moto"],"_links":{"self":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/13707","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/comments?post=13707"}],"version-history":[{"count":10,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/13707\/revisions"}],"predecessor-version":[{"id":13741,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/13707\/revisions\/13741"}],"wp:attachment":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/media?parent=13707"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/categories?post=13707"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/tags?post=13707"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}