event-sourcing-tests

Build event-sourcing tests - the given-events / when-command / then-events aggregate test, replay determinism (same events produce the same state), event-versioning + upcasting, snapshot equivalence (replay-to-N vs snapshot-at-N must agree), projection rebuild from the event log, and retroactive event correction. Per martinfowler.com EventSourcing reference. Use when an event-sourced aggregate gains a new event type or a changed payload schema, when snapshots are introduced to shorten replay, or when the event log is the audit system of record.

Install with skills.sh (any agent)

npx skills add testland/qa --skill event-sourcing-tests

View source

event-sourcing-tests

Tests for an event-sourced system verify replay determinism, snapshot equivalence, and version-evolution correctness - without them, the event log silently drifts from the rebuilt state.

When to use

Domain model is event-sourced (orders, accounts, inventory).
Audit / compliance requirements demand event log as system of record.
Adding a new event type or changing payload schema - retro-compat tests are mandatory.

How to use

Write the aggregate's given-events / when-command / then-events test first (see Worked example) - it is the base every other check builds on.
Add a replay-determinism assertion: replay the same log twice and assert identical state; this catches time.now() / random-ID leaks at the source.
When snapshots are introduced, assert snapshot-at-N equals replay-to-N - see references/snapshot-versioning-projections.md.
When an event type or payload schema changes, add versioned upcasters and a mixed-version replay test - see the references (versioning + upcasting).
For read models, assert projection rebuild from the log is idempotent and matches incremental application - see the references (projection rebuild).
Gate appends on optimistic concurrency (expected version) and suppress external calls in replay mode - see the references (concurrency, replay mode).

Worked example - one event-sourced aggregate, end to end

The base test for any event-sourced aggregate is given-events / when-command / then-events: seed the aggregate with its prior events, handle one command, then assert the events it emits and the resulting state. Per Fowler - Event Sourcing (opens in new window), replay = "rebuild application state from scratch by replaying events in order."

def test_confirm_order_emits_order_confirmed():
    # GIVEN - the aggregate's prior event history
    history = [
        OrderCreated(order_id="o1", customer="c1"),
        ItemAdded(order_id="o1", sku="sku1", qty=2),
    ]
    order = OrderAggregate.replay(history)

    # WHEN - one command is handled
    new_events = order.handle(ConfirmOrder(order_id="o1"))

    # THEN - assert the emitted events, not only the final state
    assert new_events == [OrderConfirmed(order_id="o1")]

    # AND replay is deterministic: the same log rebuilds the same state
    state_a = OrderAggregate.replay(history + new_events)
    state_b = OrderAggregate.replay(history + new_events)
    assert state_a == state_b
    assert state_a.status == "confirmed"
    assert state_a.line_items == [("sku1", 2)]

If handle emits events derived from time.now() or random IDs, state_a == state_b fails - the test catches non-deterministic replay before it drifts the log from the rebuilt state.

Once this base test is green, layer on the operational checks - order independence within causality, snapshot equivalence, event versioning + upcasting, projection rebuild, retroactive correction, replay-mode side-effect suppression, and optimistic-concurrency appends - in references/snapshot-versioning-projections.md.

Anti-patterns

Anti-pattern	Why it fails	Fix
Replay calls `time.now()` / random IDs	Non-deterministic	Make replay deterministic (Worked example)
Skip snapshot equivalence test	Snapshots silently diverge	Assert snapshot-at-N == replay-to-N (references)
No upcasting plan; rewrite event store on schema change	Audit loss; downtime	Versioned upcasters (references)
Real email/HTTP calls during replay	Duplicate side effects	Replay-mode flag (references)
Append without expected version	Lost updates from concurrent writers	Optimistic concurrency (references)

Limitations

Event-store implementations vary widely (EventStoreDB, Kafka, Postgres). Test against the actual store.
Snapshot strategy choice (every N events, every X minutes) has performance implications outside this skill's scope.
Cross-aggregate transactions are not part of event sourcing - use sagas (saga-transaction-tests) for those.

References

Fowler - Event Sourcing (opens in new window) - pattern overview, replay, snapshots, retroactive corrections, gateway considerations.
references/snapshot-versioning-projections.md - the deep operational tests: order independence, snapshot equivalence, versioning + upcasting, projection rebuild, retroactive correction, replay-mode side effects, and optimistic-concurrency appends.
saga-transaction-tests - cross-aggregate transactions.
cqrs-projection-tests - projection-from-event-log testing.

Event-sourcing tests - snapshots, versioning, projections, concurrency

View source

Event-sourcing tests - snapshots, versioning, projections, concurrency

Deep reference for the event-sourcing-tests SKILL.md. Consult after the base given-events / when-command / then-events test (in SKILL.md) is green, when adding snapshots, evolving event schemas, rebuilding projections, or hardening appends against concurrent writers.

Order independence within causality

Within a single aggregate, events ARE causally ordered. Across aggregates, only causal events are ordered. Test the boundary:

def test_unrelated_aggregates_replay_independently():
    # Two orders; events interleaved in the log
    log = [
        OrderCreated("o1", "c1"),
        OrderCreated("o2", "c2"),
        ItemAdded("o2", "sku2", 1),
        ItemAdded("o1", "sku1", 2),
        OrderConfirmed("o2"),
        OrderConfirmed("o1"),
    ]

    o1 = OrderAggregate.replay(filter(lambda e: e.order_id == "o1", log))
    o2 = OrderAggregate.replay(filter(lambda e: e.order_id == "o2", log))

    assert o1.line_items == [("sku1", 2)]
    assert o2.line_items == [("sku2", 1)]

Snapshot equivalence

Snapshots cache replayed state at version N. Per Fowler - Event Sourcing (opens in new window), "Most implementations cache the current application state, using snapshots to avoid replaying thousands of events."

Test snapshot at version N == replay-to-version N:

def test_snapshot_equivalent_to_full_replay():
    events = [...]  # 1000 events

    full_replay = OrderAggregate.replay(events)
    snapshot = OrderAggregate.snapshot_at(events, version=500)
    after_snapshot = OrderAggregate.from_snapshot(snapshot).apply_from(events[500:])

    assert full_replay == after_snapshot

If snapshot diverges, snapshot-creation logic is broken or events post-snapshot apply differently than they did during snapshot creation.

Event versioning + upcasting

Schema evolves: ItemAdded(sku, qty) → ItemAdded(sku, qty, unit_price). Old events lack unit_price - upcast on read.

def test_upcasting_v1_to_v2():
    v1_event = {"type": "ItemAdded", "version": 1, "sku": "sku1", "qty": 2}
    v2_event = upcast_v1_v2(v1_event)

    assert v2_event["version"] == 2
    assert v2_event["unit_price"] is None  # or default per business rule
    assert v2_event["sku"] == "sku1"
    assert v2_event["qty"] == 2

Test that replaying with a mix of v1 + v2 events produces the same state as if all were v2:

def test_replay_handles_mixed_event_versions():
    mixed = [v1_event, v2_event, v1_event]
    upgraded = [upcast(e) for e in mixed]

    state_via_upcast = OrderAggregate.replay(upgraded)
    assert state_via_upcast.line_items_count == 3

Projection rebuild from the event log

Read models (projections) are derived from events. Rebuild from scratch must produce the same result:

def test_projection_rebuild_idempotent():
    events = load_events()
    projection_a = build_projection(events)
    projection_b = build_projection(events)
    assert projection_a == projection_b

    # And rebuild from cleared state matches incremental update
    projection_c = SearchProjection()
    for evt in events:
        projection_c.apply(evt)
    assert projection_c == projection_a

When projection logic changes, drop the materialized view and rebuild from events - test the rebuild matches expectations.

Retroactive event correction

Per Fowler - Event Sourcing (opens in new window), "Incorrect past events can be reversed and corrected, with downstream consequences automatically recalculated."

def test_retroactive_correction_recomputes():
    # Original state: order o1 has 2 items
    events_v1 = [OrderCreated("o1"), ItemAdded("o1", "sku1", 2)]
    state_v1 = OrderAggregate.replay(events_v1)
    assert state_v1.total_items == 2

    # Discover ItemAdded was wrong (qty 5, not 2). Two strategies:
    # (a) Append correction event:
    events_v1.append(ItemQtyCorrected("o1", "sku1", new_qty=5))
    state_corrected = OrderAggregate.replay(events_v1)
    assert state_corrected.total_items == 5

    # (b) Or replace the original event in the log + replay:
    events_v2 = [OrderCreated("o1"), ItemAdded("o1", "sku1", 5)]
    state_replayed = OrderAggregate.replay(events_v2)
    assert state_replayed.total_items == 5

Strategy (a) preserves audit trail (corrections visible). Strategy (b) requires careful migration but produces a clean log. Tests verify both yield the right final state.

External system integration during replay

Per Fowler - Event Sourcing (opens in new window): "Gateways must distinguish between real processing and replay modes to avoid sending duplicate notifications or using stale data."

def test_replay_mode_suppresses_external_calls():
    email_gateway = MockEmailGateway()

    handler = OrderConfirmedHandler(email_gateway, mode="replay")
    handler.handle(OrderConfirmed("o1"))

    assert email_gateway.sent_count == 0  # replay mode = no real calls

def test_live_mode_invokes_external_calls():
    email_gateway = MockEmailGateway()

    handler = OrderConfirmedHandler(email_gateway, mode="live")
    handler.handle(OrderConfirmed("o1"))

    assert email_gateway.sent_count == 1

Optimistic concurrency on append

Append must check expected version:

def test_concurrent_append_rejected():
    events = [OrderCreated("o1")]
    store.append("o1", events, expected_version=0)  # OK; new version = 1

    # Two concurrent commands both load version=1 + try to append
    new_events_a = [ItemAdded("o1", "sku1", 1)]
    new_events_b = [ItemAdded("o1", "sku2", 1)]

    store.append("o1", new_events_a, expected_version=1)  # OK; new version = 2

    with pytest.raises(ConcurrencyConflict):
        store.append("o1", new_events_b, expected_version=1)  # already at 2

Related skills

cqrs-projection-tests

Build CQRS read-model projection tests - write-model + read-model consistency tests, projection-replay determinism, projection-versioning + zero-downtime swap, eventual-consistency-window assertions. Per martinfowler.com CQRS reference. Use when a read model is derived from a write-model event stream - adding a projection, migrating a projection schema, or chasing a "I changed it but the UI shows the old value" report.

eventual-consistency-tests

Build eventual-consistency tests for distributed infrastructure: multi-region replication convergence windows ("within 5s"), monotonic-read guarantees, anti-entropy self-healing, and CRDT merge semantics (OR-Set, G-Counter, LWW, vector clocks). Distinguishes "eventually" from "never" by asserting bounded convergence. Use when the consistency boundary is a cache cluster, replication topology, or CRDT store, not a CQRS command/query split (use cqrs-projection-tests for read-model lag after a command).

outbox-pattern-test-author

Authors tests for the transactional outbox pattern: atomic DB-write-plus-event-insert in one transaction, relay/poller publishing with at-least-once delivery and consumer deduplication, insertion-order preservation, idempotent consumers, and relay failure/retry. Use when adding outbox infrastructure, changing the relay or poller, or auditing whether dual-write atomicity and at-least-once delivery guarantees hold under failure.

saga-transaction-tests

Build saga transaction tests - orchestration vs choreography variants, per-step compensating-action verification, partial-failure scenarios (Step 3 fails → Steps 1+2 must compensate), idempotency of compensations, outbox pattern for atomic DB-update + message-publish. Per microservices.io/saga; tests guard against ACD-without-Isolation anomalies. Use when a business transaction spans two or more services and the compensating paths for a mid-sequence failure have no coverage, or right after compensation logic changes.