Chapter 5: Testing the Ledger

We have been running code manually in scripts and the Protean shell. It is time for proper automated tests. In this chapter we will discover Protean's testing DSL — a fluent API that reads like English: "given an Account after these events, process this command."

The Testing DSL

Protean provides a given() function that starts a test sentence:

from protean.testing import given

The DSL chains three concepts:

given(Aggregate, ...events) — set up aggregate history
.process(Command) — dispatch a command through the full pipeline
Assert the result — .accepted, .rejected, .events, etc.

Setting Up Tests

Create a conftest.py that initializes the domain:

# conftest.py


@pytest.fixture(autouse=True)
def fidelis_domain():
    domain.init(traverse=False)
    with domain.domain_context():
        yield domain

And create reusable event fixtures:

@pytest.fixture
def account_opened():
    """Pre-built event representing an opened account."""
    return AccountOpened(
        account_id="acc-123",
        account_number="ACC-001",
        holder_name="Alice Johnson",
        opening_deposit=1000.00,
    )


@pytest.fixture
def funded_account(account_opened):
    """Events representing an account with some transaction history."""
    return [
        account_opened,
        DepositMade(account_id="acc-123", amount=500.00, reference="paycheck"),
        DepositMade(account_id="acc-123", amount=200.00, reference="refund"),
    ]

Testing Account Creation

When no prior events exist, use given(Account) with no event arguments:

def test_open_account():
    result = given(Account).process(
        OpenAccount(
            account_number="ACC-NEW",
            holder_name="Bob Smith",
            opening_deposit=500.00,
        )
    )

    assert result.accepted
    assert AccountOpened in result.events
    assert result.events[AccountOpened].holder_name == "Bob Smith"
    assert result.holder_name == "Bob Smith"
    assert result.balance == 500.00

Key assertions:

result.accepted — the command was processed successfully.
AccountOpened in result.events — the EventLog supports in for type checking.
result.events[AccountOpened] — access the event instance by type.
result.holder_name — the result proxies attribute access to the aggregate, so you can check final state directly.

Testing with History

When an account already exists, seed the event store with prior events:

def test_deposit_increases_balance(account_opened):
    result = given(Account, account_opened).process(
        MakeDeposit(account_id="acc-123", amount=500.00, reference="paycheck")
    )

    assert result.accepted
    assert DepositMade in result.events
    assert result.events[DepositMade].amount == 500.00
    assert result.balance == 1500.00  # 1000 + 500

given(Account, account_opened) tells the DSL: "create an Account by replaying this event, then process the command." The command handler loads the aggregate (which now has a $1,000 balance from the AccountOpened event), calls deposit(), and persists.

Testing Rejections

When a command violates a business rule, use .rejected:

def test_overdraft_is_rejected(account_opened):
    result = given(Account, account_opened).process(
        MakeWithdrawal(account_id="acc-123", amount=5000.00)
    )

    assert result.rejected
    assert any("Insufficient funds" in m for m in result.rejection_messages)
    assert len(result.events) == 0

result.rejected — the command was rejected (a ValidationError was raised).
result.rejection_messages — a flat list of error strings from the ValidationError.
len(result.events) == 0 — no events were recorded because the command was rejected.

Multi-Command Chaining

Test an entire lifecycle by chaining .process() calls:

def test_full_account_lifecycle(account_opened):
    result = (
        given(Account, account_opened)
        .process(MakeDeposit(account_id="acc-123", amount=500.00))
        .process(MakeDeposit(account_id="acc-123", amount=200.00))
        .process(MakeWithdrawal(account_id="acc-123", amount=1700.00))
        .process(CloseAccount(account_id="acc-123", reason="Moving abroad"))
    )

    assert result.accepted
    assert result.balance == 0.00
    assert result.status == "CLOSED"
    assert len(result.all_events) == 4  # Deposit + Deposit + Withdrawal + Close

Each .process() builds on the state left by the previous one. .events always reflects the last command, while .all_events accumulates events across all chained commands.

Testing Invariant Violations

Verify that the "cannot close with balance" invariant works:

def test_cannot_close_with_balance(account_opened):
    result = given(Account, account_opened).process(
        CloseAccount(account_id="acc-123", reason="Customer request")
    )

    assert result.rejected
    assert any(
        "Cannot close account with non-zero balance" in m
        for m in result.rejection_messages
    )

Running the Tests

$ pytest test_fidelis.py -v
test_fidelis.py::test_open_account PASSED
test_fidelis.py::test_deposit_increases_balance PASSED
test_fidelis.py::test_overdraft_is_rejected PASSED
test_fidelis.py::test_full_account_lifecycle PASSED
test_fidelis.py::test_cannot_close_with_balance PASSED

5 passed in 0.12s

Why Integration Tests?

Notice that given().process() runs the full pipeline: it calls domain.process(), which routes to the command handler, which loads the aggregate from the event store, calls the domain method, persists the result, and returns. There are no mocks.

This is deliberate. Event-sourced systems derive their state from events — mocking the event store would defeat the purpose. The testing DSL makes integration tests fast enough (in-memory) and expressive enough (fluent API) that you rarely need unit tests with mocks.

What We Built

given(Account) for testing creation commands.
given(Account, event1, event2) for testing with prior history.
.process(Command) to dispatch through the full pipeline.
.accepted and .rejected for asserting outcomes.
EventLog with in, [], .types, .first, .last.
Multi-command chaining with .process().process().
.all_events for cross-command event accumulation.
.rejection_messages for flat error string access.

Part I is complete. We have a solid banking ledger with business rules and comprehensive tests. In Part II, we will grow the platform with projections, event handlers, async processing, and cross-aggregate coordination.

Full Source

"""Chapter 5: Testing the Ledger

This file contains the domain elements and tests for the Fidelis
banking ledger, demonstrating Protean's event-sourced testing DSL.
"""

import pytest

from protean import Domain, apply, handle, invariant
from protean.exceptions import ValidationError
from protean.fields import Float, Identifier, String
from protean.testing import given
from protean.utils.globals import current_domain

domain = Domain("fidelis")


@domain.event(part_of="Account")
class AccountOpened:
    account_id: Identifier(required=True)
    account_number: String(required=True)
    holder_name: String(required=True)
    opening_deposit: Float(required=True)


@domain.event(part_of="Account")
class DepositMade:
    account_id: Identifier(required=True)
    amount: Float(required=True)
    reference: String()


@domain.event(part_of="Account")
class WithdrawalMade:
    account_id: Identifier(required=True)
    amount: Float(required=True)
    reference: String()


@domain.event(part_of="Account")
class AccountClosed:
    account_id: Identifier(required=True)
    reason: String()


@domain.aggregate(is_event_sourced=True)
class Account:
    account_number: String(max_length=20, required=True)
    holder_name: String(max_length=100, required=True)
    balance: Float(default=0.0)
    status: String(max_length=20, default="ACTIVE")

    @invariant.post
    def balance_must_not_be_negative(self):
        if self.balance is not None and self.balance < 0:
            raise ValidationError(
                {"balance": ["Insufficient funds: balance cannot be negative"]}
            )

    @invariant.post
    def closed_account_must_have_zero_balance(self):
        if self.status == "CLOSED" and self.balance != 0:
            raise ValidationError(
                {"status": ["Cannot close account with non-zero balance"]}
            )

    @classmethod
    def open(cls, account_number: str, holder_name: str, opening_deposit: float):
        account = cls._create_new()
        account.raise_(
            AccountOpened(
                account_id=str(account.id),
                account_number=account_number,
                holder_name=holder_name,
                opening_deposit=opening_deposit,
            )
        )
        return account

    def deposit(self, amount: float, reference: str = None) -> None:
        if amount <= 0:
            raise ValidationError({"amount": ["Deposit amount must be positive"]})
        self.raise_(
            DepositMade(account_id=str(self.id), amount=amount, reference=reference)
        )

    def withdraw(self, amount: float, reference: str = None) -> None:
        if amount <= 0:
            raise ValidationError({"amount": ["Withdrawal amount must be positive"]})
        self.raise_(
            WithdrawalMade(account_id=str(self.id), amount=amount, reference=reference)
        )

    def close(self, reason: str = None) -> None:
        self.raise_(AccountClosed(account_id=str(self.id), reason=reason))

    @apply
    def on_account_opened(self, event: AccountOpened):
        self.id = event.account_id
        self.account_number = event.account_number
        self.holder_name = event.holder_name
        self.balance = event.opening_deposit
        self.status = "ACTIVE"

    @apply
    def on_deposit_made(self, event: DepositMade):
        self.balance += event.amount

    @apply
    def on_withdrawal_made(self, event: WithdrawalMade):
        self.balance -= event.amount

    @apply
    def on_account_closed(self, event: AccountClosed):
        self.status = "CLOSED"


@domain.command(part_of=Account)
class OpenAccount:
    account_number: String(required=True)
    holder_name: String(required=True)
    opening_deposit: Float(required=True)


@domain.command(part_of=Account)
class MakeDeposit:
    account_id: Identifier(required=True)
    amount: Float(required=True)
    reference: String()


@domain.command(part_of=Account)
class MakeWithdrawal:
    account_id: Identifier(required=True)
    amount: Float(required=True)
    reference: String()


@domain.command(part_of=Account)
class CloseAccount:
    account_id: Identifier(required=True)
    reason: String()


@domain.command_handler(part_of=Account)
class AccountCommandHandler:
    @handle(OpenAccount)
    def handle_open_account(self, command: OpenAccount):
        account = Account.open(
            account_number=command.account_number,
            holder_name=command.holder_name,
            opening_deposit=command.opening_deposit,
        )
        current_domain.repository_for(Account).add(account)
        return str(account.id)

    @handle(MakeDeposit)
    def handle_make_deposit(self, command: MakeDeposit):
        repo = current_domain.repository_for(Account)
        account = repo.get(command.account_id)
        account.deposit(command.amount, reference=command.reference)
        repo.add(account)

    @handle(MakeWithdrawal)
    def handle_make_withdrawal(self, command: MakeWithdrawal):
        repo = current_domain.repository_for(Account)
        account = repo.get(command.account_id)
        account.withdraw(command.amount, reference=command.reference)
        repo.add(account)

    @handle(CloseAccount)
    def handle_close_account(self, command: CloseAccount):
        repo = current_domain.repository_for(Account)
        account = repo.get(command.account_id)
        account.close(reason=command.reason)
        repo.add(account)


# conftest.py


@pytest.fixture(autouse=True)
def fidelis_domain():
    domain.init(traverse=False)
    with domain.domain_context():
        yield domain


@pytest.fixture
def account_opened():
    """Pre-built event representing an opened account."""
    return AccountOpened(
        account_id="acc-123",
        account_number="ACC-001",
        holder_name="Alice Johnson",
        opening_deposit=1000.00,
    )


@pytest.fixture
def funded_account(account_opened):
    """Events representing an account with some transaction history."""
    return [
        account_opened,
        DepositMade(account_id="acc-123", amount=500.00, reference="paycheck"),
        DepositMade(account_id="acc-123", amount=200.00, reference="refund"),
    ]




def test_open_account():
    result = given(Account).process(
        OpenAccount(
            account_number="ACC-NEW",
            holder_name="Bob Smith",
            opening_deposit=500.00,
        )
    )

    assert result.accepted
    assert AccountOpened in result.events
    assert result.events[AccountOpened].holder_name == "Bob Smith"
    assert result.holder_name == "Bob Smith"
    assert result.balance == 500.00


def test_deposit_increases_balance(account_opened):
    result = given(Account, account_opened).process(
        MakeDeposit(account_id="acc-123", amount=500.00, reference="paycheck")
    )

    assert result.accepted
    assert DepositMade in result.events
    assert result.events[DepositMade].amount == 500.00
    assert result.balance == 1500.00  # 1000 + 500


def test_overdraft_is_rejected(account_opened):
    result = given(Account, account_opened).process(
        MakeWithdrawal(account_id="acc-123", amount=5000.00)
    )

    assert result.rejected
    assert any("Insufficient funds" in m for m in result.rejection_messages)
    assert len(result.events) == 0


def test_full_account_lifecycle(account_opened):
    result = (
        given(Account, account_opened)
        .process(MakeDeposit(account_id="acc-123", amount=500.00))
        .process(MakeDeposit(account_id="acc-123", amount=200.00))
        .process(MakeWithdrawal(account_id="acc-123", amount=1700.00))
        .process(CloseAccount(account_id="acc-123", reason="Moving abroad"))
    )

    assert result.accepted
    assert result.balance == 0.00
    assert result.status == "CLOSED"
    assert len(result.all_events) == 4  # Deposit + Deposit + Withdrawal + Close


def test_cannot_close_with_balance(account_opened):
    result = given(Account, account_opened).process(
        CloseAccount(account_id="acc-123", reason="Customer request")
    )

    assert result.rejected
    assert any(
        "Cannot close account with non-zero balance" in m
        for m in result.rejection_messages
    )