Testing

Testing Event Sourcing and CQRS: Aggregates, Projections, and Event Handlers

30 Sep 2026 — 4 min read

Event sourcing is a different mental model for state. Instead of reading and writing a current state, you store a sequence of events and derive current state by replaying them. This has testing implications: you test the events that commands produce, not the state mutations directly.

The Testing Model

In event-sourced systems:

Commands express intent (CreateOrder, ShipOrder)
Aggregates apply commands and produce events
Events are the source of truth (OrderCreated, OrderShipped)
Projections read events and build query-optimized views

The test pattern for aggregates follows Given-When-Then:

Given: a sequence of past events (the aggregate's history)
When: a command is applied
Then: new events are produced (or an error is raised)

Testing Aggregates: Pure Event Logic

# order_aggregate.py
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime

@dataclass
class OrderCreated:
    order_id: str
    customer_id: str
    items: List[dict]
    timestamp: datetime = field(default_factory=datetime.utcnow)

@dataclass
class OrderShipped:
    order_id: str
    tracking_number: str
    timestamp: datetime = field(default_factory=datetime.utcnow)

@dataclass
class OrderCancelled:
    order_id: str
    reason: str
    timestamp: datetime = field(default_factory=datetime.utcnow)

class Order:
    def __init__(self):
        self._order_id = None
        self._customer_id = None
        self._status = None
        self._items = []
        self._events = []  # uncommitted events
    
    @classmethod
    def from_events(cls, events):
        """Reconstitute from event history"""
        order = cls()
        for event in events:
            order._apply(event)
        return order
    
    def create(self, order_id, customer_id, items):
        if not items:
            raise ValueError("Order must have at least one item")
        event = OrderCreated(order_id=order_id, customer_id=customer_id, items=items)
        self._apply(event)
        self._events.append(event)
    
    def ship(self, tracking_number):
        if self._status != 'pending':
            raise ValueError(f"Cannot ship order in status: {self._status}")
        event = OrderShipped(order_id=self._order_id, tracking_number=tracking_number)
        self._apply(event)
        self._events.append(event)
    
    def cancel(self, reason):
        if self._status == 'shipped':
            raise ValueError("Cannot cancel a shipped order")
        event = OrderCancelled(order_id=self._order_id, reason=reason)
        self._apply(event)
        self._events.append(event)
    
    def _apply(self, event):
        if isinstance(event, OrderCreated):
            self._order_id = event.order_id
            self._customer_id = event.customer_id
            self._items = event.items
            self._status = 'pending'
        elif isinstance(event, OrderShipped):
            self._status = 'shipped'
        elif isinstance(event, OrderCancelled):
            self._status = 'cancelled'
    
    def get_uncommitted_events(self):
        return list(self._events)
    
    def mark_events_committed(self):
        self._events.clear()

# test_order_aggregate.py
import pytest
from order_aggregate import Order, OrderCreated, OrderShipped, OrderCancelled

class TestOrderCreation:
    def test_create_produces_order_created_event(self):
        order = Order()
        order.create(order_id='ord-001', customer_id='cust-1', items=[{'sku': 'A', 'qty': 2}])
        
        events = order.get_uncommitted_events()
        
        assert len(events) == 1
        assert isinstance(events[0], OrderCreated)
        assert events[0].order_id == 'ord-001'
        assert events[0].customer_id == 'cust-1'
    
    def test_create_with_empty_items_raises(self):
        order = Order()
        
        with pytest.raises(ValueError, match="at least one item"):
            order.create(order_id='ord-001', customer_id='cust-1', items=[])

class TestOrderShipping:
    def test_ship_pending_order_produces_shipped_event(self):
        # Given: an existing pending order (reconstituted from history)
        history = [
            OrderCreated(order_id='ord-001', customer_id='cust-1', items=[{'sku': 'A'}])
        ]
        order = Order.from_events(history)
        
        # When: ship command
        order.ship(tracking_number='TRACK123')
        
        # Then: OrderShipped event produced
        events = order.get_uncommitted_events()
        assert len(events) == 1
        assert isinstance(events[0], OrderShipped)
        assert events[0].tracking_number == 'TRACK123'
    
    def test_cannot_ship_cancelled_order(self):
        history = [
            OrderCreated(order_id='ord-001', customer_id='cust-1', items=[{'sku': 'A'}]),
            OrderCancelled(order_id='ord-001', reason='customer request')
        ]
        order = Order.from_events(history)
        
        with pytest.raises(ValueError, match="cannot ship"):
            order.ship(tracking_number='TRACK123')
    
    def test_cannot_ship_already_shipped_order(self):
        history = [
            OrderCreated(order_id='ord-001', customer_id='cust-1', items=[{'sku': 'A'}]),
            OrderShipped(order_id='ord-001', tracking_number='TRACK123')
        ]
        order = Order.from_events(history)
        
        with pytest.raises(ValueError, match="pending"):
            order.ship(tracking_number='TRACK456')

The key: aggregates have no database dependency. from_events reconstructs state from a list of plain Python objects. Tests are pure unit tests — fast, no mocking needed.

Testing Projections

Projections read events and build read models. Test them by feeding events and asserting on the read model.

# order_projection.py
class OrderSummaryProjection:
    def __init__(self):
        self._summaries = {}
    
    def on_order_created(self, event):
        self._summaries[event.order_id] = {
            'order_id': event.order_id,
            'customer_id': event.customer_id,
            'item_count': len(event.items),
            'status': 'pending'
        }
    
    def on_order_shipped(self, event):
        if event.order_id in self._summaries:
            self._summaries[event.order_id]['status'] = 'shipped'
            self._summaries[event.order_id]['tracking_number'] = event.tracking_number
    
    def on_order_cancelled(self, event):
        if event.order_id in self._summaries:
            self._summaries[event.order_id]['status'] = 'cancelled'
    
    def get_summary(self, order_id):
        return self._summaries.get(order_id)
    
    def get_all_by_status(self, status):
        return [s for s in self._summaries.values() if s['status'] == status]

def test_projection_builds_summary_from_events():
    projection = OrderSummaryProjection()
    
    projection.on_order_created(
        OrderCreated(order_id='ord-001', customer_id='cust-1', items=[{'sku': 'A'}, {'sku': 'B'}])
    )
    
    summary = projection.get_summary('ord-001')
    
    assert summary['status'] == 'pending'
    assert summary['item_count'] == 2

def test_projection_updates_on_shipped_event():
    projection = OrderSummaryProjection()
    
    projection.on_order_created(
        OrderCreated(order_id='ord-001', customer_id='cust-1', items=[{'sku': 'A'}])
    )
    projection.on_order_shipped(
        OrderShipped(order_id='ord-001', tracking_number='TRACK123')
    )
    
    summary = projection.get_summary('ord-001')
    assert summary['status'] == 'shipped'
    assert summary['tracking_number'] == 'TRACK123'

def test_projection_filters_by_status():
    projection = OrderSummaryProjection()
    
    projection.on_order_created(OrderCreated(order_id='ord-001', customer_id='c1', items=[{'sku': 'A'}]))
    projection.on_order_created(OrderCreated(order_id='ord-002', customer_id='c2', items=[{'sku': 'B'}]))
    projection.on_order_shipped(OrderShipped(order_id='ord-001', tracking_number='T1'))
    
    pending = projection.get_all_by_status('pending')
    shipped = projection.get_all_by_status('shipped')
    
    assert len(pending) == 1
    assert pending[0]['order_id'] == 'ord-002'
    assert len(shipped) == 1
    assert shipped[0]['order_id'] == 'ord-001'

Java with Axon Framework

Axon provides a AggregateTestFixture that implements the Given-When-Then pattern natively.

@ExtendWith(MockitoExtension.class)
class OrderAggregateTest {
    
    private AggregateTestFixture<OrderAggregate> fixture;
    
    @BeforeEach
    void setUp() {
        fixture = new AggregateTestFixture<>(OrderAggregate.class);
    }
    
    @Test
    void createOrderProducesOrderCreatedEvent() {
        fixture
            .givenNoPriorActivity()
            .when(new CreateOrderCommand("ord-001", "cust-1", List.of(new OrderItem("SKU-A", 2))))
            .expectSuccessfulHandlerExecution()
            .expectEvents(
                new OrderCreatedEvent("ord-001", "cust-1", List.of(new OrderItem("SKU-A", 2)))
            );
    }
    
    @Test
    void shipOrderAfterCreationProducesShippedEvent() {
        fixture
            .given(new OrderCreatedEvent("ord-001", "cust-1", List.of(new OrderItem("SKU-A", 1))))
            .when(new ShipOrderCommand("ord-001", "TRACK123"))
            .expectSuccessfulHandlerExecution()
            .expectEvents(new OrderShippedEvent("ord-001", "TRACK123"));
    }
    
    @Test
    void shippingCancelledOrderThrowsException() {
        fixture
            .given(
                new OrderCreatedEvent("ord-001", "cust-1", List.of(new OrderItem("SKU-A", 1))),
                new OrderCancelledEvent("ord-001", "customer request")
            )
            .when(new ShipOrderCommand("ord-001", "TRACK123"))
            .expectException(IllegalStateException.class)
            .expectExceptionMessage(containsString("cannot ship"));
    }
}

Axon's AggregateTestFixture handles event store setup automatically. You just provide the history in given() and assert on new events in expectEvents().

Testing Event Handler Idempotency

Event sourcing systems often process events multiple times (at-least-once delivery). Test that your handlers are idempotent.

def test_projection_handles_duplicate_events_idempotently():
    projection = OrderSummaryProjection()
    
    created_event = OrderCreated(
        order_id='ord-001',
        customer_id='cust-1',
        items=[{'sku': 'A'}]
    )
    
    # Apply the same event twice (simulating at-least-once delivery)
    projection.on_order_created(created_event)
    projection.on_order_created(created_event)  # duplicate
    
    # Should still have exactly one summary
    all_summaries = list(projection._summaries.values())
    assert len(all_summaries) == 1

Event sourcing makes this test natural — just apply the event twice and check the result. If your projection isn't idempotent, this test catches it.

The Event Sourcing Testing Advantage

One underappreciated benefit of event sourcing: tests are trivial to reproduce. Any bug in production becomes:

Capture the event stream that led to the bug
Replay it in a test
Assert on the broken behavior
Fix it
The test captures the regression

No database snapshots, no complex state setup. Just a list of events.

Testing Event Sourcing and CQRS: Aggregates, Projections, and Event Handlers

The Testing Model

Testing Aggregates: Pure Event Logic

Testing Projections

Java with Axon Framework

Testing Event Handler Idempotency

The Event Sourcing Testing Advantage

Read more

Table-Driven Tests in Go: Patterns and Best Practices

Swift Mocking Patterns Using Protocols

Mocking in Go with testify/mock: A Practical Guide

Swift Package Testing with SPM: Unit Tests, Code Coverage, and Xcode Cloud