Unit test DynamoDB in Python with pytest and dynalite

Introduction

Recently I was working on a little project recently with Python and Chalice, a serverless framework Python. I had a logic error in one of my DynamoDB queries that I didn't catch until I deployed the project. Because I was new to using DynamoDB with Python, it took a few more deployments over an hour before I fixed the query.

Do you know what would've been nice? Having access to DynamoDB locally so I could have written that query using Test-Driven Development (TDD).

Why couldn't I? Well, code used in serverless development, like with AWS Lambda, isn't designed to run on your local machine:

Serverless is a cloud-native development model that allows developers to build and run applications without having to manage servers. (Red Hat: "What is serverless?")

The "cloud-native" descriptor is important here. With serverless development, and Function-as-a-Service (FaaS) development specifically in this case, it's possible to publish a function without, say, having to set up a webserver, configure a reverse proxy, and add SSL. The servers are still there, of course, but the server management is abstracted away by the cloud hosting provider.

Serverless development being cloud-native is a double-edged sword, however, because although you can build applications without administering servers by leveraging cloud development frameworks, your applications are now designed to run in a cloud environment.

What does this mean? For one, it can be difficult to run your applications on your local machine. What does that mean? Testing complicated queries in a cloud-native database like DynamoDB can be error-prone and cumbersome.

There are services to help with local serverless development, like LocalStack, and frameworks that make cloud development feel more like local development, like Serverless Stack, but LocalStack was overkill for me and I didn't want to refactor my code to work with Serverless Stack. I just needed a way to test functions containing DynamoDB queries since I could test the rest of my application using mocks.

That's where dynalite comes in.

Setup

To unit test DynamoDB locally, we're going to use dynalite, an in-memory implementation of DynamoDB, along with boto3 for Python bindings to DynamoDB, and pytest to run the unit tests.

First, install dynalite globally:

npm install -g dynalite

or locally:

npm install -D dynalite

Then, install pytest and boto3, the AWS Python SDK:

poetry add boto3 pytest

Pytest config

Now that we have our dependencies installed, let's tie things together.

To use dynalite in our unit tests, we'll need to ensure that dynalite is running before our unit tests start. We could start dynalite in a separate terminal window before running pytest and stop that dynalite process after the tests finish, but that sounds like a lot of work. Instead, we can use a pytest fixture in a file called conftest.py:

# conftest.py
import subprocess

import boto3
from pytest import fixture

TABLE_NAME = "PersonTestTable"
DYNALITE_PORT = "4567"


@fixture(scope="session")
def dynamodb_table(request):
    proc = subprocess.Popen(
        ["dynalite", "--port", DYNALITE_PORT, "--createTableMs", "0"]
    )
    dynamodb = boto3.resource(
        "dynamodb",
        endpoint_url=f"http://localhost:{DYNALITE_PORT}",
    )

    request.addfinalizer(lambda: proc.kill())

    try:
        table = dynamodb.create_table(
            TableName=TABLE_NAME,
            KeySchema=[
                {"AttributeName": "PK", "KeyType": "HASH"},
                {"AttributeName": "SK", "KeyType": "RANGE"},
            ],
            AttributeDefinitions=[
                {"AttributeName": "PK", "AttributeType": "S"},
                {"AttributeName": "SK", "AttributeType": "S"},
            ],
            BillingMode="PAY_PER_REQUEST",
        )
        yield table
    except Exception:
        yield dynamodb.Table(TABLE_NAME)

This feature does a few things.

First, it's scoped to the pytest session so it will be available to every test that runs in that pytest invocation:

@fixture(scope="session")
def dynamodb_table(request):

Second, because this fixture runs dynalite in a subprocess, we need to stop that subprocess once the tests have run. To do that, the fixture includes a "finalizer", a callable that contains cleanup/teardown code (docs):

# proc is the subprocess running dynalite
request.addfinalizer(lambda: proc.kill())

Third, the fixture creates a DynamoDB table:

table = dynamodb.create_table(
    TableName=TABLE_NAME,
    KeySchema=[
        {"AttributeName": "PK", "KeyType": "HASH"},
        {"AttributeName": "SK", "KeyType": "RANGE"},
    ],
    AttributeDefinitions=[
        {"AttributeName": "PK", "AttributeType": "S"},
        {"AttributeName": "SK", "AttributeType": "S"},
    ],
    BillingMode="PAY_PER_REQUEST",
)

Finally, with what is admitedly a dirty hack, the fixture creates the table only once by wrapping the dynamodb.create_table() call in a try/except block that returns the table if there's an error creating it:

def dynamodb_table(request):
    # ...
    try:
        table = dynamodb.create_table(
            # arguments omitted for brevity
        )
        yield table
    except Exception:
        yield dynamodb.Table(TABLE_NAME)

Example

Example test using fixture:

def test_find_people_with_middle_initials(dynamodb_table):
    dynamodb_table.put_item(
        Item=dict(
            PK="Person",
            SK="123456",
            first_name="John",
            middle_initial="R",
            last_name="Franey",
        )
    )
    dynamodb_table.put_item(
        Item=dict(
            PK="Person",
            SK="234567",
            first_name="Bilbo",
            last_name="Baggins",
        )
    )
    people_with_middle_initials = find_people_with_middle_initials()
    assert len(people_with_middle_initials) == 1
    assert people_with_middle_initials[0].get("middle_initial") == "R"

Wrap-up

That's that! With a guest appearance from the Node.js world, dynalite, it's easy to unit test DynamoDB queries in Python on your local machine.