Event Sourcing Testing: How to Test Event-Sourced Systems Correctly

30 Sep 2026 — 6 min read

Event sourcing inverts the traditional data model — instead of storing current state, you store a sequence of events that describe what happened. This makes testing both easier and harder. Easier because events are immutable facts you can inspect directly. Harder because bugs in event replay logic corrupt all derived state, not just the current row.

This guide covers practical patterns for testing event-sourced systems.

The Given/When/Then Testing Pattern

Event sourcing has a natural testing vocabulary:

Given: a sequence of past events that set up aggregate state
When: a command is handled
Then: new events are emitted (or an error is raised)

This maps directly to test structure:

class TestOrderAggregate:
    def test_confirm_order_emits_order_confirmed_event(self):
        """
        Given: An order has been placed
        When: The order is confirmed
        Then: An OrderConfirmed event is emitted
        """
        # Given
        order = Order.reconstitute([
            OrderPlaced(order_id="o1", customer_id="c1", total=99.99)
        ])
        
        # When
        events = order.confirm(confirmed_by="admin-1")
        
        # Then
        assert len(events) == 1
        assert isinstance(events[0], OrderConfirmed)
        assert events[0].order_id == "o1"
        assert events[0].confirmed_by == "admin-1"

    def test_cannot_confirm_already_confirmed_order(self):
        """
        Given: An order has been placed and confirmed
        When: Someone tries to confirm it again
        Then: An error is raised, no events emitted
        """
        # Given
        order = Order.reconstitute([
            OrderPlaced(order_id="o1", customer_id="c1", total=99.99),
            OrderConfirmed(order_id="o1", confirmed_by="admin-1")
        ])
        
        # When / Then
        with pytest.raises(InvalidOperationError, match="already confirmed"):
            order.confirm(confirmed_by="admin-2")

    def test_cancel_after_shipment_is_rejected(self):
        """
        Given: Order placed, confirmed, and shipped
        When: Cancellation is attempted
        Then: Error raised — cannot cancel shipped orders
        """
        order = Order.reconstitute([
            OrderPlaced(order_id="o1", customer_id="c1", total=99.99),
            OrderConfirmed(order_id="o1", confirmed_by="admin-1"),
            OrderShipped(order_id="o1", tracking_number="TRK123")
        ])
        
        with pytest.raises(InvalidOperationError, match="cannot cancel.*shipped"):
            order.cancel(reason="customer request")

Testing Aggregate Reconstitution

The reconstitute method (applying past events to rebuild state) is critical — a bug here corrupts all future state:

class TestOrderReconstitution:
    def test_reconstitute_builds_correct_state_from_events(self):
        """Aggregate state is correctly reconstructed from event sequence."""
        events = [
            OrderPlaced(order_id="o1", customer_id="c1", total=99.99, items=["item-1"]),
            ItemAdded(order_id="o1", item_id="item-2", price=29.99),
            OrderConfirmed(order_id="o1", confirmed_by="admin-1"),
        ]
        
        order = Order.reconstitute(events)
        
        assert order.id == "o1"
        assert order.customer_id == "c1"
        assert order.status == OrderStatus.CONFIRMED
        assert len(order.items) == 2
        assert order.total == 129.98  # 99.99 + 29.99

    def test_empty_event_stream_creates_uninitialized_aggregate(self):
        """Reconstituting with no events creates an uninitialized aggregate."""
        order = Order.reconstitute([])
        assert order.id is None
        assert order.status == OrderStatus.UNINITIALIZED

    def test_reconstitution_is_deterministic(self):
        """Reconstituting the same events always produces the same state."""
        events = [
            OrderPlaced(order_id="o1", customer_id="c1", total=99.99),
            OrderConfirmed(order_id="o1", confirmed_by="admin-1"),
        ]
        
        order1 = Order.reconstitute(events)
        order2 = Order.reconstitute(events)
        
        assert order1.status == order2.status
        assert order1.total == order2.total

    def test_out_of_order_events_raise_error(self):
        """Events applied out of sequence are rejected."""
        events = [
            # Confirmed before placed — invalid sequence
            OrderConfirmed(order_id="o1", confirmed_by="admin-1"),
            OrderPlaced(order_id="o1", customer_id="c1", total=99.99),
        ]
        
        with pytest.raises(InvalidEventSequenceError):
            Order.reconstitute(events)

Testing Projections (Read Models)

Projections build read models from event streams. Test that they handle all event types correctly:

class TestOrderProjection:
    def test_projection_builds_order_summary(self):
        """Projection creates correct read model from events."""
        projection = OrderSummaryProjection()
        
        projection.apply(OrderPlaced(
            order_id="o1", customer_id="c1",
            total=99.99, placed_at=datetime(2024, 1, 1, 10, 0)
        ))
        projection.apply(OrderConfirmed(
            order_id="o1", confirmed_by="admin-1",
            confirmed_at=datetime(2024, 1, 1, 10, 5)
        ))
        
        summary = projection.get("o1")
        
        assert summary.order_id == "o1"
        assert summary.customer_id == "c1"
        assert summary.status == "confirmed"
        assert summary.total == 99.99
        assert summary.placed_at == datetime(2024, 1, 1, 10, 0)
        assert summary.confirmed_at == datetime(2024, 1, 1, 10, 5)

    def test_projection_handles_cancellation(self):
        """Projection correctly updates read model when order is cancelled."""
        projection = OrderSummaryProjection()
        
        projection.apply(OrderPlaced(order_id="o1", customer_id="c1", total=50.0))
        projection.apply(OrderCancelled(order_id="o1", reason="customer request"))
        
        summary = projection.get("o1")
        assert summary.status == "cancelled"
        assert summary.cancellation_reason == "customer request"

    def test_projection_ignores_unknown_events(self):
        """Unknown event types are ignored without error."""
        projection = OrderSummaryProjection()
        
        # Should not raise
        projection.apply(UnknownFutureEvent(order_id="o1"))
        
        # Summary should not exist for unknown-only events
        assert projection.get("o1") is None

    def test_projection_is_idempotent(self):
        """Applying the same event twice produces the same result."""
        projection = OrderSummaryProjection()
        event = OrderPlaced(order_id="o1", customer_id="c1", total=99.99)
        
        projection.apply(event)
        projection.apply(event)  # Replay/duplicate
        
        # Count should still be 1 (idempotent)
        assert len(projection.get_all()) == 1

Testing Event Replay

Event replay (rebuilding state from scratch by replaying all events) is the most critical operation in event sourcing. Any bug makes all derived state wrong:

class TestEventReplay:
    def test_full_replay_produces_correct_final_state(self, event_store, projection_store):
        """Full event replay produces identical state to incremental projection."""
        # Build state incrementally (normal operation)
        events = [
            OrderPlaced(order_id="o1", customer_id="c1", total=99.99),
            OrderConfirmed(order_id="o1", confirmed_by="admin-1"),
            OrderShipped(order_id="o1", tracking_number="TRK123"),
        ]
        
        for event in events:
            event_store.append(event)
            projection_store.apply(event)
        
        incremental_state = projection_store.get("o1")
        
        # Reset projection and replay from event store
        projection_store.reset()
        projection_store.replay(event_store.get_all())
        
        replayed_state = projection_store.get("o1")
        
        assert replayed_state.status == incremental_state.status
        assert replayed_state.total == incremental_state.total
        assert replayed_state.tracking_number == incremental_state.tracking_number

    def test_replay_handles_schema_evolution(self, event_store):
        """Old event format is correctly upcasted to current schema during replay."""
        # Simulate an old event with deprecated field name
        old_format_event = {
            "type": "OrderPlaced",
            "version": 1,
            "order_id": "o1",
            "amount": 99.99,  # old field name, now called "total"
        }
        
        event_store.append_raw(old_format_event)
        
        # Replay should upcast and handle correctly
        projection = OrderSummaryProjection()
        projection.replay(event_store.get_all())
        
        summary = projection.get("o1")
        assert summary.total == 99.99  # Correctly upcasted from "amount"

Testing Command Handlers

Command handlers validate commands and dispatch them to aggregates:

class TestPlaceOrderCommandHandler:
    def test_place_order_creates_order_placed_event(self, event_store, repository):
        """PlaceOrder command results in OrderPlaced event in event store."""
        handler = PlaceOrderCommandHandler(repository, event_store)
        
        command = PlaceOrderCommand(
            order_id="o1",
            customer_id="c1",
            items=[{"product_id": "p1", "quantity": 2, "price": 49.99}]
        )
        
        handler.handle(command)
        
        events = event_store.get_events("o1")
        assert len(events) == 1
        assert isinstance(events[0], OrderPlaced)
        assert events[0].order_id == "o1"
        assert events[0].total == 99.98  # 2 * 49.99

    def test_place_order_with_invalid_customer_raises_error(self, event_store, mock_customer_service):
        """PlaceOrder with nonexistent customer is rejected."""
        mock_customer_service.exists.return_value = False
        
        handler = PlaceOrderCommandHandler(
            repository=...,
            event_store=event_store,
            customer_service=mock_customer_service
        )
        
        command = PlaceOrderCommand(order_id="o1", customer_id="nonexistent", items=[])
        
        with pytest.raises(CustomerNotFoundError):
            handler.handle(command)
        
        # No events should be stored
        events = event_store.get_events("o1")
        assert events == []

    def test_place_order_is_idempotent(self, event_store, repository):
        """Handling the same PlaceOrder command twice produces only one event."""
        handler = PlaceOrderCommandHandler(repository, event_store)
        
        command = PlaceOrderCommand(
            order_id="o1",
            customer_id="c1",
            items=[{"product_id": "p1", "quantity": 1, "price": 99.99}]
        )
        
        handler.handle(command)
        handler.handle(command)  # Duplicate (network retry simulation)
        
        events = event_store.get_events("o1")
        assert len(events) == 1, "Duplicate command should not create duplicate event"

Testing Snapshots

Snapshots speed up aggregate loading by storing state at a point in time. Test that snapshot creation and restoration work correctly:

class TestOrderSnapshot:
    def test_snapshot_captures_current_state(self, repository):
        """Snapshot correctly captures aggregate state at a point in time."""
        order = Order.reconstitute([
            OrderPlaced(order_id="o1", customer_id="c1", total=99.99),
            OrderConfirmed(order_id="o1", confirmed_by="admin-1"),
        ])
        
        snapshot = order.take_snapshot()
        
        assert snapshot.aggregate_id == "o1"
        assert snapshot.version == 2
        assert snapshot.state["status"] == "confirmed"
        assert snapshot.state["total"] == 99.99

    def test_restore_from_snapshot_then_apply_new_events(self, event_store, snapshot_store):
        """Aggregate can be restored from snapshot and then apply subsequent events."""
        # Create snapshot after 100 events
        snapshot = Snapshot(
            aggregate_id="o1",
            version=100,
            state={"status": "confirmed", "total": 99.99}
        )
        snapshot_store.save(snapshot)
        
        # Add new events after snapshot
        event_store.append(OrderShipped(
            order_id="o1",
            tracking_number="TRK123",
            version=101
        ))
        
        # Load from snapshot + new events
        order = repository.load("o1")
        
        assert order.status == OrderStatus.SHIPPED
        assert order.tracking_number == "TRK123"
        assert order.version == 101

CI/CD Integration

Event sourcing tests are typically fast (no external dependencies for unit tests):

# .github/workflows/event-sourcing-tests.yml
name: Event Sourcing Tests

on: [push, pull_request]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install pytest
      - run: pytest tests/aggregates/ tests/projections/ tests/commands/ -v

  integration-tests:
    runs-on: ubuntu-latest
    services:
      eventstore:
        image: eventstore/eventstore:23.10.0-jammy
        ports:
          - 2113:2113
        env:
          EVENTSTORE_CLUSTER_SIZE: 1
          EVENTSTORE_RUN_PROJECTIONS: All
          EVENTSTORE_INSECURE: true
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install pytest esdbclient
      - run: pytest tests/integration/ -v
        env:
          EVENTSTORE_URL: esdb://localhost:2113?tls=false

Common Event Sourcing Testing Mistakes

Testing only the happy path: Event sourcing bugs often appear in edge cases — concurrent commands, events applied in the wrong order, replaying events with old schemas.

Not testing idempotency: Command handlers and projections should be idempotent — processing the same event twice should produce the same result as processing it once.

No schema evolution tests: Event schemas change over time. Test that old events are correctly upcasted to the current schema during replay.

Testing implementation details: Don't assert on the internal state of the aggregate — assert on the events emitted. The events are the API.

No concurrency tests: Concurrent commands on the same aggregate cause optimistic concurrency conflicts. Test that your optimistic locking works correctly.

Event sourcing systems have a natural testing structure — given/when/then maps directly to reconstitute/command/events. The patterns here (aggregate tests, projection tests, replay correctness, command handler idempotency) give you the coverage to catch event sourcing bugs before they corrupt your event store.