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.