Testing Event-Driven Systems

Tests at several levels

Most event-driven logic is ordinary code, so test it as such. Unit tests: call handlers and aggregates directly with events and check resulting state or emitted events; event-sourced aggregates read naturally as given these past events, when this command, then these new events. Idempotency tests: deliver the same event twice and assert the effect happened once. Contract tests: check that producers still emit events matching the schema consumers rely on, using a schema registry's compatibility checks or consumer-driven contract tools such as Pact's message support. Integration tests: run a real broker and database in containers (Testcontainers) to check serialisation, outbox relays and consumer configuration. End-to-end tests of whole flows are slow and flaky; keep a few for critical paths and use polling with timeouts (eventually) rather than fixed sleeps.

Given-when-then for an event-sourced aggregate

History in, command applied, new events out.

def test_cannot_withdraw_more_than_balance():
    # given
    acct = Account("a-7")
    for e in [{"type": "MoneyDeposited", "amount": 1000}]:
        acct.apply(e)
    # when / then
    with pytest.raises(ValueError):
        acct.withdraw(1500)
    assert acct.pending == []

def test_payment_captured_is_idempotent(db):
    event = make_event("PaymentCaptured", order_id="o-1001")
    handle_payment_captured(event, db)
    handle_payment_captured(event, db)          # duplicate delivery
    assert db.count("payments_applied", order_id="o-1001") == 1

Testing a relay race

You time each runner alone (unit tests), check that the baton fits every hand (contract tests), and run a few full races before the competition (end-to-end tests). Running only full races tells you something is slow, but not who.

Quick check: Why are contract tests important for events?

• They make brokers faster
• They replace unit tests
• They test the network
• They catch producer changes that would break consumers before deployment