Testing¶

waku provides testing utilities for event-sourced aggregates and deciders. The DeciderSpec DSL enables expressive Given/When/Then tests for functional deciders.

DeciderSpec DSL¶

DeciderSpec provides a fluent Given/When/Then API for testing IDecider implementations.

The basic chain is:

DeciderSpec.for_(decider).given([events]).when(command).then([expected_events])

from app.decider import (
    BankAccountDecider,
    DepositMoney,
    OpenAccount,
)
from app.events import AccountOpened, MoneyDeposited
from waku.eventsourcing.testing import DeciderSpec


def test_open_account() -> None:
    decider = BankAccountDecider()

    (
        DeciderSpec
        .for_(decider)
        .given([])
        .when(OpenAccount(account_id='acc-1', owner='dex'))
        .then([AccountOpened(account_id='acc-1', owner='dex')])
    )


def test_deposit_updates_balance() -> None:
    decider = BankAccountDecider()

    (
        DeciderSpec
        .for_(decider)
        .given([AccountOpened(account_id='acc-1', owner='dex')])
        .when(DepositMoney(account_id='acc-1', amount=500))
        .then([MoneyDeposited(account_id='acc-1', amount=500)])
    )


def test_deposit_negative_raises() -> None:
    decider = BankAccountDecider()

    (
        DeciderSpec
        .for_(decider)
        .given([AccountOpened(account_id='acc-1', owner='dex')])
        .when(DepositMoney(account_id='acc-1', amount=-10))
        .then_raises(ValueError, match='Deposit amount must be positive')
    )


def test_state_after_events() -> None:
    decider = BankAccountDecider()

    (
        DeciderSpec
        .for_(decider)
        .given([
            AccountOpened(account_id='acc-1', owner='dex'),
            MoneyDeposited(account_id='acc-1', amount=500),
        ])
        .when(DepositMoney(account_id='acc-1', amount=300))
        .then_state(lambda s: s.balance == 800)
    )

DeciderSpec Methods¶

These methods set up the test scenario. given() is optional — omit it to test from initial state.

Method	Parameters	Returns	Description
`for_`	`decider: IDecider[S, C, E]`	`DeciderSpec[S, C, E]`	Class method. Create a spec for the given decider
`given`	`events: Sequence[E]`	`DeciderSpec[S, C, E]`	Apply prior events to build up state before the command
`when`	`command: C`	`_DeciderWhenResult[S, C, E]`	Execute a command against the built-up state
`then_state`	`predicate: Callable[[S], bool]`	`None`	Assert state built from `given()` events alone (no command)

Assertions After `.when(command)`¶

Available after .when(command):

Method	Parameters	Returns	Description
`then`	`expected_events: Sequence[E]`	`None`	Assert the command produced exactly these events
`then_no_events`	—	`None`	Assert the command produced zero events
`then_raises`	`exception_type: type[Exception]`, `match: str \| None = None`	`None`	Assert the command raises this exception. `match` is a regex passed to `pytest.raises`
`then_state`	`predicate: Callable[[S], Any]`	`None`	Assert the state after applying produced events matches the predicate
`resulting_state`	—	`S`	Property. Returns the state after deciding and evolving — use for custom assertions

Tip

then_state appears on both DeciderSpec and the result of .when(). On DeciderSpec it checks state from events alone (no command). After .when() it checks state after the command's produced events are applied.

OOP Aggregate Testing¶

The pattern for testing OOP aggregates: create the aggregate, optionally call load_from_history() to set up prior state, invoke a command method, then assert collect_events() and state.

from app.aggregate import BankAccount
from app.events import AccountOpened, MoneyDeposited


def test_aggregate_opens_account() -> None:
    account = BankAccount()
    account.open('acc-1', 'dex')

    events = account.collect_events()
    assert len(events) == 1
    assert isinstance(events[0], AccountOpened)
    assert events[0].owner == 'dex'


def test_aggregate_deposits_money() -> None:
    account = BankAccount()
    account.load_from_history(
        [AccountOpened(account_id='acc-1', owner='dex')],
        version=0,
    )

    account.deposit('acc-1', 500)

    events = account.collect_events()
    assert len(events) == 1
    assert isinstance(events[0], MoneyDeposited)
    assert account.balance == 500

Tip

load_from_history() lets you set up any starting state without going through the full event store.

Integration Testing¶

For integration tests, use InMemoryEventStore (the default) — no database needed. Combine it with waku.testing.create_test_app() to create minimal test applications.

from waku.cqrs import IMediator
from waku.testing import create_test_app

from app.commands import OpenAccountCommand
from app.modules import AppModule


async def test_full_flow() -> None:
    async with create_test_app(base=AppModule) as app:
        async with app.container() as container:
            mediator = await container.get(IMediator)
            result = await mediator.send(OpenAccountCommand(account_id='acc-1', owner='dex'))
            assert result.account_id == 'acc-1'

Tip

The default EventSourcingConfig() already uses in-memory stores, so no extra configuration is needed for tests.