6/27/2016

DynamoDB in pytest-dbfixtures

pytest-dbfixtures

If you use pytest maybe you have heard about pytest-dbfixtures:
Pytest dbfixtures is a pytest plugin that makes it a lot easier to set up proper database or storage engine for testing. Simply use one of provided fixtures that start predefined clean database server for your tests or creates server more tailored for your application by using one of provided factories.
This plugin is very useful if you have integration tests in your project, and you want to perform tests on a database for example. You will find information how to use it in the documentation.

Currently, the plugin supports:
  • Postgresql
  • MySQL
  • Redis
  • Mongo
  • Elasticsearch
  • RabbitMQ
And recently, we have added support for DynamoDB.

Here, you will find how to run DynamoDB on your computer. And, here we are. If you want to use it in production, you want to test it locally.

dynamodb fixture

If you still do not use pytest, go to the pytest page and read how to use it, now. If you want to understand the further content you have to know how pytest works.

The simplest way is to put dynamodb fixture as an argument in your test:
1
2
def test_dynamodb(dynamodb):
    ...

You will find an instance of the class DynamoDB.ServiceResource in dynamodb variable. The plugin assumes you have extracted DynamoDB files into /tmp/dynamodb. If you have not extracted the files into this directory, or you do not want to do this, you can create an individual fixture:
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from pytest_dbfixtures import factories

my_dynamodb_proc = factories.dynamodb_proc(
    dynamodb_dir="/path/to/mydynamodb",
    port=14235,
    delay=True,
)
my_dynamodb = factories.dynamodb("my_dynamodb_proc")


def test_my_dynamodb(my_dynamodb):
    ...

As you can see, you may also pass port and delay as arguments. If you do not specify the port argument, the fixture will use a random port. If you want to introduce delays for certain operations in DynamoDB you can pass the argument delay as a True.
DynamoDB can perform some tasks almost instantaneously, such as create/update/delete operations on tables and indexes; however, the actual DynamoDB service requires more time for these tasks. Setting this parameter helps DynamoDB simulate the behavior of the Amazon DynamoDB web service more closely. (Currently, this parameter introduces delays only for global secondary indexes that are in either CREATING or DELETING status.)
So, let's say that you want to run DynamoDB on 9009 port with some delays, and you keep DynamoDB files in /tmp/tests/dynamodb. Then you want to create a table and put some data there.

Here is an example:
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from pytest_dbfixtures import factories

my_dynamodb_proc = factories.dynamodb_proc(
    dynamodb_dir="/tmp/tests/dynamodb",
    port=9009,
    delay=True,
)
my_dynamodb = factories.dynamodb("my_dynamodb_proc")


def test_my_dynamodb(my_dynamodb):
    table = my_dynamodb.create_table(
        TableName="Test",
        KeySchema=[
            {
                "AttributeName": "id",
                "KeyType": "HASH"
            }
        ],
        AttributeDefinitions=[
            {
                "AttributeName": "id",
                "AttributeType": "N"
            }

        ],
        ProvisionedThroughput={
            "ReadCapacityUnits": 10,
            "WriteCapacityUnits": 10
        }
    )

    table.put_item(
        Item={
            "id": 42,
            "my_data": "is secret"
        }
    )

    response = table.get_item(
        Key={
            "id": 42,
        }
    )

    assert response["Item"]["my_data"] == "is secret"

Because we want to test faster and faster, we run DynamoDB in memory. Remember that when you stop the process of DynamoDB, none of the data will be saved (DynamoDB does not write any database files at all). If you do not want to run tests in memory, your Pull Request is welcome.