Testing Event-Sourced Applications

Hello! Welcome back to our course on distributed systems architecture.

In our last lesson, we built a complete, albeit simple, CQRS system. You implemented a BankingApplication (the write model) and an AccountsProjection that populates a queryable AccountsView (the read model). This separation is powerful, but it also introduces complexity and multiple moving parts. How can we be confident that our aggregate's business rules are correctly enforced, or that our projection accurately reflects the event stream?

This lesson directly addresses that question by focusing on our current learning outcome:

Write unit and integration tests for an event-sourced application using the framework's testing utilities.

We will explore a structured approach to testing our event-sourced system, covering three distinct levels:

1. Unit testing the Account aggregate's business logic in isolation.
2. Integration testing the BankingApplication service to ensure it correctly persists events.
3. Integration testing the full projection pipeline to verify that events from the write model correctly update the read model.

To do this, we'll use pytest, a powerful and popular Python testing framework.

1. The Testing Toolkit: pytest

While you're a solid Python programmer, it's worth establishing a common foundation for our testing framework. pytest simplifies test creation by using plain assert statements and a powerful fixture system for managing test setup and teardown.

First, let's install the necessary libraries:

pip install pytest pytest-mock

pytest-mock is a plugin that makes it easy to "mock" or fake parts of our system, which is essential for writing isolated unit tests.

To get a quick but thorough introduction to the key features we'll be using, please watch the following video.

2. Unit Testing the Aggregate: Given-When-Then

The heart of an event-sourced system is the aggregate, which contains the core business logic. Unit testing the aggregate means verifying this logic in complete isolation, without involving any databases or application services.

The standard pattern for this is Given-When-Then:

• Given a history of past events...
• When a command is executed...
• Then new events are produced, OR an exception is thrown if a business rule is violated.

This pattern maps perfectly to how aggregates work: they replay past events to reconstruct state and then process new commands.

Implementation

Let's apply this to our Account aggregate. First, create a tests directory and a file inside it named test_domain.py.

.
├── banking_app.py
├── projections.py
└── tests/
    └── test_domain.py

Now, add the following code to test_domain.py.

# tests/test_domain.py
from decimal import Decimal
import pytest
from uuid import uuid4

from banking_app import Account, InsufficientFundsError

def test_account_deposit():
    # GIVEN an opened account
    account_opened = Account.Opened(
        originator_id=uuid4(),
        originator_version=0,
        owner="Alice",
        initial_balance=Decimal("100"),
    )
    account = Account.__reconstitute__(event_stream=(account_opened,))
    assert account.balance == Decimal("100")

    # WHEN we deposit funds
    account.deposit(Decimal("50"))

    # THEN a FundsDeposited event is raised and the balance is updated
    pending_events = account.collect_events()
    assert len(pending_events) == 1
    
    deposited_event = pending_events[0]
    assert isinstance(deposited_event, Account.FundsDeposited)
    assert deposited_event.amount == Decimal("50")
    assert account.balance == Decimal("150") # The aggregate state is updated immediately

def test_withdraw_insufficient_funds():
    # GIVEN an account with a balance of 100
    account_opened = Account.Opened(
        originator_id=uuid4(),
        originator_version=0,
        owner="Alice",
        initial_balance=Decimal("100"),
    )
    account = Account.__reconstitute__(event_stream=(account_opened,))

    # WHEN we try to withdraw more than the balance
    # THEN an InsufficientFundsError is raised
    with pytest.raises(InsufficientFundsError):
        account.withdraw(Decimal("150"))

    # AND no events are raised
    assert len(account.collect_events()) == 0

Key Points:

• We use the internal __reconstitute__ method to create an aggregate instance from a history of events (the "Given" part).
• We call a command method like deposit() or withdraw() (the "When" part).
• We use account.collect_events() to check the newly generated events and assert to check the aggregate's state (the "Then" part).
• We use pytest.raises() to verify that our business rule invariants (like not overdrawing an account) are correctly enforced.

3. Integration Testing the Write Model

Unit tests are great for business logic, but they don't tell us if our application can correctly save and retrieve aggregates from the event store. An integration test for the BankingApplication service fills this gap.

The eventsourcing library makes this straightforward by providing an in-memory persistence module. We can run our tests against this module without needing a real database, making them fast and isolated.

The synopsis on the library's main documentation page provides a perfect template for this kind of test.

Implementation

Create a new file, tests/test_application.py. We'll use a pytest fixture to provide a clean BankingApplication instance for each test.

# tests/test_application.py
import os
from decimal import Decimal
import pytest

from banking_app import BankingApplication, Account

@pytest.fixture
def banking_app() -> BankingApplication:
    """Provides a fresh, in-memory BankingApplication for each test."""
    # Set env var for in-memory persistence, which is the default but good to be explicit
    os.environ["PERSISTENCE_MODULE"] = "eventsourcing.popo"
    app = BankingApplication()
    return app

def test_open_account_and_deposit(banking_app: BankingApplication):
    # Execute commands against the application service
    account_id = banking_app.open_account(owner="Bob", initial_balance=Decimal("50"))
    banking_app.deposit_funds(account_id, Decimal("25"))

    # Retrieve the aggregate from the repository to verify state
    account: Account = banking_app.repository.get(account_id)
    
    # Assert the final state is correct
    assert account.owner == "Bob"
    assert account.balance == Decimal("75")

def test_withdraw_from_account(banking_app: BankingApplication):
    # Setup
    account_id = banking_app.open_account(owner="Charlie", initial_balance=Decimal("200"))

    # Execute command
    banking_app.withdraw_funds(account_id, Decimal("70"))

    # Verify
    account: Account = banking_app.repository.get(account_id)
    assert account.balance == Decimal("130")

The @pytest.fixture ensures that test_open_account_and_deposit and test_withdraw_from_account run independently, each with a brand new, empty BankingApplication instance. This prevents tests from interfering with each other.

4. Integration Testing the Projection (End-to-End)

This is the most comprehensive test. It verifies that an event created by the write model is successfully processed by our projection, resulting in a correct update to the read model. This tests the entire CQRS flow.

We can achieve this by running both the BankingApplication and the ProjectionRunner within the same test function, using the library's in-memory notification system.

Implementation

Create a final test file, tests/test_projection.py.

# tests/test_projection.py
import os
from decimal import Decimal
import pytest
from typing import Tuple

from eventsourcing.projection import ProjectionRunner

from banking_app import BankingApplication
from projections import AccountsProjection, AccountsView, IAccountsView

@pytest.fixture
def in_memory_cqrs_system() -> Tuple[BankingApplication, IAccountsView]:
    """
    Sets up a full in-memory CQRS system with a write app and a read view.
    The ProjectionRunner runs synchronously in-process for the test.
    """
    # Configure both app and projection to use in-memory storage
    os.environ["PERSISTENCE_MODULE"] = "eventsourcing.popo"
    os.environ["ACCOUNTS_PERSISTENCE_MODULE"] = "eventsourcing.popo"
    
    # The ProjectionRunner connects the application to the projection
    runner = ProjectionRunner(
        application_class=BankingApplication,
        projection_class=AccountsProjection,
        view_class=AccountsView,
    )
    
    # The runner holds instances of the app and the view
    write_model: BankingApplication = runner.application
    read_model: IAccountsView = runner.view
    
    # The 'with' statement starts the runner's processing
    with runner:
        yield write_model, read_model
    # The runner is stopped automatically on exit from 'with'

def test_projection_updates_on_account_opened(
    in_memory_cqrs_system: Tuple[BankingApplication, IAccountsView]
):
    write_model, read_model = in_memory_cqrs_system

    # Check that the read model is initially empty
    assert read_model.get_accounts() == []

    # WHEN we open an account via the write model
    account_id = write_model.open_account(owner="Denise", initial_balance=Decimal("1000"))

    # THEN the read model is updated
    accounts = read_model.get_accounts()
    assert len(accounts) == 1
    assert accounts[0]["account_id"] == account_id
    assert accounts[0]["owner"] == "Denise"
    assert accounts[0]["balance"] == Decimal("1000")

def test_projection_updates_on_deposit(
    in_memory_cqrs_system: Tuple[BankingApplication, IAccountsView]
):
    write_model, read_model = in_memory_cqrs_system

    # GIVEN an existing account
    account_id = write_model.open_account(owner="Eve", initial_balance=Decimal("200"))

    # WHEN we deposit funds
    write_model.deposit_funds(account_id, Decimal("50"))

    # THEN the read model's balance is updated
    accounts = read_model.get_accounts()
    assert len(accounts) == 1
    assert accounts[0]["balance"] == Decimal("250")

To run all your tests, navigate to the root directory of your project in the terminal and simply execute the pytest command.

pytest

pytest will automatically discover and run all files named test_*.py and all functions named test_* within them.

Conclusion

In this lesson, you've learned a comprehensive, multi-layered strategy for testing an event-sourced application. This approach gives you high confidence in your system's correctness, from the finest-grained business rule to the end-to-end data flow.

Key Takeaways:

• Unit Tests for Aggregates: Use the Given-When-Then pattern to test business logic in isolation. Reconstitute aggregates from events, execute a command, and assert on the outcome (new events or an exception).
• Integration Tests for Services: Use pytest fixtures and the eventsourcing library's in-memory persistence (popo) to test application services. This verifies the command-handling and persistence logic without external dependencies.
• End-to-End Projection Tests: The most powerful tests verify the entire CQRS flow. By running the Application and ProjectionRunner together in-memory, you can assert that write-side events correctly update your read models.
• pytest is your friend: Fixtures are essential for creating isolated, repeatable test environments.

Preview of the Next Lesson:

Our application is now well-structured and thoroughly tested. The next step is to prepare it for a real, distributed deployment. In the upcoming lesson, "Implement health checks and readiness probes for a distributed Python service," we will begin to explore the operational aspects of running distributed systems, ensuring they are observable and manageable in a production environment.