db-delta 1.0.3

Python library to evaluate DynamoDB changesets when running database unittests.

DB Delta

db-delta is a Python testing utility package designed to be used alongside moto that evaluates and validates database changesets in tests that execute AWS DynamoDB operations to ensure that only the changes that you define are executed against your database, no more and no less.

Usage is simple.

Step 1: Define a changeset in JSON format:

[
    {
        "change_type": "updated_item",
        "key":{
            "PK": "FOO",
            "SK": "BAR"
        },
        "updated_fields": [
            {
                "field": "bar",
                "update_type": "updated",
                "new_value": "bar-foo"
            },
            {
                "field": "foo",
                "update_type": "removed",
            }
        ]
    }
]

Step 2: Execute test function with the validate_dynamodb_changeset context manager

from db_delta import ChangeSet, validate_dynamodb_changeset

def test_function_foo(table):

    expected_changeset = ChangeSet.from_json("path_to_changeset.json")

    with validate_dynamodb_changeset(table, expected_changeset):
        # call your function that modifies the contents of your database.
        # db_delta will ensure that only the changes specified are executed.
        execute_function(table, "foo", "bar")

db_delta will scan your DynamoDB table before and after the execution of your function. It will then verify that only the changes defined in the changeset have been executed. If any of the specified changes are not found, or if any changes have been made that are not defined in the changeset, an AssertionException is raised.

Why use db_delta?

Writing exhaustive unittests to test database operations is just that. Exhausting. Unittests end up being either incomplete, or extremely verbose and repetitive. db_delta not ony makes tests far more readable, it also makes them far more robust. If any change is made to any item in your database that has not been defined in the changeset, your tests will fail. No exceptions and no surprises.

Consider the following example code.

def example_function(table, new_name: str):
    table.update_item(
        Key={
            "PK": "FOO",
            "SK": "BAR",
        },
        UpdateExpression="SET #name = :name",
        ExpressionAttributeNames={
            "#name": "name",
        },
        ExpressionAttributeValues={
            ":name": new_name,
        },
    )

A typical unittest would look like the following.

def test_example_function(table):

    item = table.get_item(
        Key={
            "PK": "FOO",
            "SK": "BAR",
        },
    )["Item"]
    assert item["name"] == "name"

    example_function(table, "name V2")

    item = table.get_item(
        Key={
            "PK": "FOO",
            "SK": "BAR",
        },
    )["Item"]
    assert item["name"] == "name V2"

There are two issues with the above.

1. Tests like the above quickly become extremely repetitive, with a great deal of duplicated code to read from your DynamoDB table and compare the difference in responses.
2. The test only checks if one field in one item has been updated as expected. Every other item from the database could have been deleted and the test would still pass.

The second point is critical and can lead to serious issues in your data. While the above example is trivial, real-life codebases often have more complex functions that execute multiple database operations in multiple places, and it quickly becomes difficult to test everything without writing long unittests that check for every possible outcome.

This is where db_delta comes in. Because it scans your table before and after the function you are testing is executed, it generates a complete picture of the changes made to your database as the result of running the code you are testing. It can then compare the initial and final state of your table to make sure that only the changes you expect have actually been executed.

It also neatens up the code. With db_delta, the above test now becomes

def test_example_function(table):

    expected_changeset = ChangeSet(
        changes=[
            {
                "change_type": "updated_item",
                "key": {
                    "PK": "FOO",
                    "SK": "BAR",
                },
                "updated_fields": [
                    {
                        "field": "name",
                        "update_type": "updated",
                        "new_value": "name V2"
                    }
                ]
            }
        ]
    )

    with validate_dynamodb_changeset(table, expected_changeset):
        example_function(table, "name V2")

The test no longer contains any logic to retrieve and validate changes from your database. Additionally, if any other changes are made other than the specified update, validate_dynamodb_changeset will raise an exception and the test will fail.

Installation

db_delta is available on PyPi and can be installed using

$ pip install db-delta

Alternatively, you can clone the repo and install the package manually with

$ pip install .

Usage

The only two components required to use db_delta is the ChangeSet object and the validate_dynamodb_changeset function.

ChangeSet Object

The ChangeSet object is a pydantic data model that contains the database changes that you expect your function to produce. Expected changes are provided as an array of objects collectively referred to as a changeset. Each object within the changeset has a change_type field indicating the type of change. Currently, db_delta supports three distinct change_type values

• put_item - represent a new item inserted into your table using a PutItem operation.
• updated_item - represents a change to an existing item using a UpdateItem operation.
• deleted_item - represents an item deleted from your table using a DeleteItem operation.

Along with change_type, each update contains a key attribute that tells db_delta what hash and range key the update is referring to.

For example, if you have a function that creates a new item and updates an existing item, you can represent the result of that function with a changeset that looks like the following:

[
	{
		"change_type": "put_item",
		 "key":{
            "PK": "BAR",
            "SK": "FOO"
        },
        "item": {
	        "PK": "BAR",
            "SK": "FOO",
            "bar": foo
        }
	},
    {
        "change_type": "updated_item",
        "key":{
            "PK": "FOO",
            "SK": "BAR"
        },
        "updated_fields": [
            {
                "field": "bar",
                "update_type": "updated",
                "new_value": "bar-foo"
            },
            {
                "field": "foo",
                "update_type": "removed",
            }
        ]
    }
]

change_type: put_item

put_item represents a new item in your database. When a changeset contains a change_type: put_item update, db_delta will check that the item did not exist in the initial table state, but does exist in the final state, and that the created item matches the expected structure. The JSON structure for a put_item update is given below.

{
    "change_type": "put_item",
    "key": {
        "foo": "bar",
        "bar": "foo"
    },
    "item": {
        "foo": "bar",
        "bar": "foo",
        "id": 1,
        "description": "An example"
    }
}

Note that the item attribute must contain the entire item, including the hash and sort key.

change_type: updated_item

updated_item represents an item in your database that has been updated. When a changeset contains a change_type: updated_item update, db_delta will check that the item exists in both the initial table state and in the final state, and that the changes made to the item match the changes defined in the changeset. The JSON structure for an updated_item update is given below.

{
    "change_type": "updated_item",
    "key": {
        "foo": "bar",
        "bar": "foo"
    },
    "updated_fields": [
        {
            "field": "newField",
            "update_type": "added",
            "new_value": "bar-foo"
        },
        {
            "field": "updatedField",
            "update_type": "updated",
            "new_value": "bar-foo"
        },
        {
            "field": "removedField",
            "update_type": "removed",
        }
    ]
}

All expected updates to the item must be defined in the updated_fields column, where each update has the following structure

{
    "field": "String",
    "update_type": "added|updated|removed",
    "new_value": "String | Integer | Float | Object"
}

The field parameter specifies the attribute of the item that the update is referring to, while the update_type parameter indicates what type of change is expected. There are three possible values for the update_type parameter

• added- an attribute that has been created as the result of the UpdatedItem operation. It must NOT exist in the initial item.
• updated - an attribute that existed in both the initial and final item, but was updated to a new value.
• removed - an attribute that was present in the initial item, but was removed using a REMOVE :key statement within your DynamoDB update. It MUST exist in the initial item.

The new_value parameter is required when update_type is one of [added,updated] and must match the final value of the specified field. new_value can be omitted if the update_type is removed.

change_type: deleted_item

deleted_item represents an item that has been deleted from your database. When a changeset contains a DeletedItem update, db_delta will check that the item exists in the initial table state, but does not exist in the final state.

{
    "change_type": "deleted_item",
    "key": {
        "foo": "bar",
        "bar": "foo"
    },
}

Generating ChangeSet instance

A ChangeSet instance can be generated one of two ways. Either define your changeset in code

expected_changeset = ChangeSet(
    changes=[
        {
            "change_type": "updated_item",
            "key": {
                "PK": "FOO",
                "SK": "BAR",
            },
            "updated_fields": [
                {
                    "field": "name",
                    "update_type": "updated",
                    "new_value": "name V2"
                }
            ]
        }
    ]
)

or define your changeset in a JSON file and use the from_json class method

expected_changeset = ChangeSet.from_json("path_to_changeset.json")

The from_json method is generally recommended as it keeps unittests smaller and allows you to re-use predefined changesets.

validate_dynamodb_changeset Function

Once you have a ChangeSet instance constructed with your expected changes, simply execute your function in the validate_dynamodb_changeset context manager.

from db_delta import ChangeSet, validate_dynamodb_changeset


def test_function_foo(table):

    expected_changeset = ChangeSet.from_json("path_to_changeset.json")

    with validate_dynamodb_changeset(table, expected_changeset):
        # call your function that modifies the contents of your database.
        # db_delta will ensure that only the changes specified are executed.
        execute_function(table, "foo", "bar")

If execute_function results in any changes that are not defined in the changeset, validate_dynamodb_changeset will raise an AssertionException.

Note that the validate_dynamodb_changeset requires both a configured DynamoDB table resource as well as the expected changeset. The table resource should be created using boto3.

table = boto3.resource("dynamodb").Table("example-table")

Test Setup

In order to get the most out of db_delta its critical that you set up your unittests correctly. db_delta is designed to be used alongside moto and pytest. We strongly recommend that you get familiar with both before integrating db_delta into your unittests. For a basic example, see the examples/basic folder in this repository, which shows you how to configure a basic testing environment with the required tools.

Advanced Usage

validate_dynamodb_changeset accepts an optional formatting function that is called on all items before making comparisons. This is especially useful if your code adds certain metadata fields such as created and update timestamps that you want to exclude from any comparisons.

For example:

def strip_metadata(row: Dict) -> Dict:
    for key in ["createdTs", "updatedTs"]:
        row.pop(key)
    return row


def test_function_foo(table):

    expected_changeset = ChangeSet.from_json("path_to_changeset.json")

    with validate_dynamodb_changeset(table, expected_changeset, formatter=strip_metadata):
        # call your function that modifies the contents of your database.
        # db_delta will ensure that only the changes specified are executed.
        execute_function(table, "foo", "bar")

The strip_metadata function is called on all items before any comparisons are made. In this case, both the createdTs and updatedTs timestamps are removed from any items, ensuring that your changesets do not need to include the updated values for mocked timestamps. This can help to keep the changesets compact.