Do you want to test the function that uses AWS resources like DynamoDB? The Python moto module is very useful while patching AWS resources. This post will take you through, how we can mock DynamoDB with the help of moto. Here, we will be mocking a method that stores input data into a given DynamoDB table. The idea would be the same for mocking other AWS resources like SQS etc.
Why do we need mocking
You can skip this section if you are familiar with the term mock. Mock is something like making a simulation, replica, or copy of anything. Let’s suppose you have to test a method, inside which another function is calling for fetching few data from somewhere else. So, we can’t afford that much cost for fetching the data every time for different test cases. Hence, we use mocking, where we can provide different data directly, without fetching it in real. Similarly, there can be more conditions for using mocking as well.
Method that we want to test
This is a simplest method that is used to write data into AWS DynamoDB table. This method takes data input, table name and then write the given data to the given DynamoDB table. For testing the proper working of the method, we don’t want to mess with our real DynamoDB table of the application. Hence, we will be creating a dummy table by mocking DynamoDB and will verify our method is working proper or not.
def write_into_table(item, table_name): dynamodb = boto3.resource('dynamodb') table = dynamodb.Table(table_name) with table.batch_writer() as batch: batch.put_item(Item=item) |
Let us assume this method is in a file called store_data.py
.
Using moto’s @mock_dynamodb2
The Python moto module is super easy to use for mocking. It provides the @mock_dynamodb2 decorator that mocks out DynamoDB. For writing this one test, we will be using the following steps:
1. Decorate the test method with @mock_dynamodb2
2. Create a DynamoDB resource
3. Create a dummy DynamoDB table
4. Create inputs data for store_data.write_into_table()
5. Call store_data.write_into_table()
6. Read the corresponding DynamoDB table
7. Verify that the retrieved data matches the supplied data
1. Decorate the test method with @mock_dynamodb2
Create a new file called test_write_into_table.py
and add the following lines:
import boto3 from moto import mock_dynamodb2 import store_data @mock_dynamodb2 def test_write_into_table(): "Test the write_into_table with a valid input data" |
Here, we are ready with the test script skeleton by importing necessary modules & decorating our test method with @mock_dynamodb2.
2. Create a DynamoDB resource
Now, within the test_write_into_table()
method, create an DynamoDB resource like following
dynamodb = boto3.resource('dynamodb') |
3. Create a dummy DynamoDB table
You can create a DynamoDB table using the DynamoDB resource like this
table_name = 'test' table = dynamodb.create_table(TableName=table_name, KeySchema=[{'AttributeName': 'date','KeyType': 'HASH'}], AttributeDefinitions=[{'AttributeName': 'date','AttributeType': 'S'}]) |
Here we created a table named ‘test’ with primary key ‘date’ of type string(‘S’)
4. Create inputs for store_data.write_into_table()
Here, we will create an input data to store_data.write_into_table()
, which will also acts as the expected data after reading.
data = {'date':'07-Oct-2020','company':'qxf2 services','client':1000} |
For understanding purpose, we took single data but one can have more as per the requirement. After writing the above data into table, the table will look like as follows:
date | company | client |
---|---|---|
07-Oct-2020 | qxf2 services | 1000 |
5. Call store_data.write_into_table()
At this point, when store_data.write_into_table() is executed, it should store the given input data to the corresponding given table.
store_data.write_into_table(data,table_name) |
6. Read the DynamoDB for data
Let’s read the data by passing ‘date’ as the primary key. Note that we will gonna get a dictionary with a couple of keys, so for getting our data we have to fetch the value of ‘Item’ key.
response = table.get_item(Key={'date':data['date']}) actual_output = response['Item'] |
7. Verify that the retrieved data matches the supplied data
Let’s assert that the body of the data is same as we stored
assert actual_output == data |
Combining everything
Here is how our test looks like :
""" Example of using moto to mock out DynamoDB table """ import boto3 from moto import mock_dynamodb2 import store_data @mock_dynamodb2 def test_write_into_table(): "Test the write_into_table with a valid input data" dynamodb = boto3.resource('dynamodb') table_name = 'test' table = dynamodb.create_table(TableName=table_name, KeySchema=[{'AttributeName': 'date','KeyType': 'HASH'}], AttributeDefinitions=[{'AttributeName': 'date','AttributeType': 'S'}]) data = {'date':'07-Oct-2020','company':'qxf2 services','client':1000} store_data.write_into_table(data,table_name) response = table.get_item(Key={'date':data['date']}) actual_output = response['Item'] assert actual_output == data |
Here, we have completed with the simplest approach to use moto with AWS DynamoDB. In similar manner one can use moto for mocking other AWS resources like SQS etc. I hope this post helps you to understand, the concept of mocking DynamoDB with moto.
I am a final year B.tech student in computer science stream. I have recently completed an Internship as a curriculum developer at Notchup, where the key responsibilities were to develop & maintain python curriculum plans from scratch. Currently doing internship as a Software Engineer Intern at Qxf2 Services. Here I learnt some of the trending tech like AWS, Automation & testing. I found Qxf2, as a great place to learn & explore for beginners like me. I fond of competitive coding, Cricket, Football, music & delicious foods.
hi Rohit,
like ur stuff, can u share the github public repo if it is available
Hi Supratik,
Please find a PR link for the same, which is under review.
https://github.com/qxf2/qxf2-lambdas/pull/8/files#diff-6dd55a2fb5c507f5552be391b68b263871d73ef86134aa095b2e640cdd07e567
Also, You could mock the DynamoDB table resource locally and unit test, provided you have a local instance of DynamoDB running.
The code change to do so would be while creating the DynamoDB resouce:
dynamodb = boto3.resource(‘dynamodb’, endpoint_url=’http://localhost:8000′)