Testing Processors

Thanks to the separation between registering and invoking processors, testing every component of the system is extremely easy. Essentially, since the processor decorator does not modify the function it decorates, it’s possible to test processors the same way any other function would be tested.

Because of the dependency injection, it’s also very easy to mock clients during testing.

Example

Suppose that we have the following functions we want to test:

from dataclasses import dataclass
from typing import Any, Dict

from event_processor import EventProcessor


event_processor = EventProcessor()


class FakeDynamoClient:
    database = {
        "users": [
            {"Email": {"S": "user@example.com"}, "Role": {"S": "user"}},
            {"Email": {"S": "admin@example.com"}, "Role": {"S": "admin"}}
        ]
    }

    def get_item(self, TableName="", Key={}):
        table = self.database.get(TableName, {})
        key_name = list(Key.keys())[0]
        record = [e for e in table if e[key_name]["S"] == Key[key_name]["S"]][0]
        return {"Item": record}


@dataclass
class User:
    email: str
    role: str


@event_processor.dependency_factory
def boto_clients(client_name: str) -> FakeDynamoClient:
    if client_name == "dynamodb":
        return FakeDynamoClient()
    else:
        raise NotImplementedError()


# Uses the dynamodb client specified in the processor decorator
def event_to_user(event: Dict, dynamodb_client: FakeDynamoClient):
    email = event["user"]["email"]
    response = dynamodb_client.get_item(
                    TableName="users",
                    Key={"Email": {"S": email}}
               )
    role = response["Item"]["Role"]["S"]

    return User(email=email, role=role)


# Does not use the dynamodb client, but needs it for pre-processing
@event_processor.processor(
    {"user.email": Any},
    pre_processor=event_to_user,
    boto_clients=("dynamodb",)
)
def my_processor(user: User):
    return user.role == "admin"


print(
    event_processor.invoke({"user": {"email": "user@example.com"}}),
    event_processor.invoke({"user": {"email": "admin@example.com"}})
)
False True

We could write the following tests:

from unittest.mock import MagicMock, patch


def test_my_processor_returns_true_for_admin_user():
    test_user = User(email="test@example.com", role="admin")

    result = my_processor(test_user)

    assert result is True


def test_event_to_user_returns_user_data_from_dynamodb():
    dynamodb_client = MagicMock()
    dynamodb_client.get_item.return_value = {
        "Item": {
            "Role": {"S": "mock-value"}
        }
    }
    test_event = {"user": {"email": "test@example.com"}}

    result = event_to_user(test_event, dynamodb_client)

    assert result.role == "mock-value"
    dynamodb_client.get_item.assert_called_once()


@patch(FakeDynamoClient)
def test_boto_clients_creates_boto_client(dynamo_client_mock):
    test_client_name = "mock-value"

    result = boto_clients(test_client_name)

    assert result == dynamo_client_mock.return_value

As you can see, the dependency injection makes the processor and pre-processor easy to test, and it makes those tests clearer by avoiding excessive patching. Patching is needed to test the dependency factory, but since that’s the only thing to test, it doesn’t make the test any less clear.