Test Event-Driven Flows End-to-End

The Problem

Unit testing individual domain elements is straightforward. You construct an aggregate, call a method, assert the resulting state. The Testing Domain Logic in Isolation pattern covers this well. But real event-driven systems are not made of isolated pieces. They are made of chains:

PlaceOrder (command)
  -> OrderCommandHandler (processes command)
    -> Order.place() (mutates aggregate, raises OrderPlaced)
      -> InventoryEventHandler (reserves stock, raises InventoryReserved)
        -> OrderDashboardProjector (updates read model)

Testing the full chain is harder. The command fires a handler, which persists an aggregate, which raises an event, which triggers another handler, which updates a projection. Each link depends on the previous one completing correctly. Without guidance, teams fall into predictable traps:

Trap 1: Skipping E2E tests entirely. "Our unit tests pass, so the chain works." It doesn't. A misconfigured stream category or missing handler registration only shows up when the full chain runs.

Trap 2: Testing handlers in isolation with fabricated events. A developer constructs an OrderPlaced event manually and calls the handler directly. The handler works, but in production the event has different metadata or is routed to a different stream.

Trap 3: Fragile async test setups. The team reaches for the Engine but wrestles with timing:

# fragment
# FRAGILE: How long do we sleep? Too short and the test is flaky.
# Too long and the suite is slow.
def test_order_flow(self):
    domain.process(PlaceOrder(...))
    time.sleep(2)  # Hope the engine processes everything...
    time.sleep(3)  # ...maybe add more time for slow CI...
    projection = domain.repository_for(OrderDashboard).get("ord-123")
    assert projection.status == "placed"  # Fails intermittently

The result: teams either avoid testing event chains entirely or build unreliable tests that erode trust in the suite.

---

The Pattern

Test event-driven flows at three levels. Each level trades speed for realism. Most teams need all three, but the middle level covers the majority of integration needs.

Level 1: Domain unit tests

Test aggregates, value objects, and domain services directly. No handlers, no repositories, no event processing infrastructure.

Aggregate -> call method -> assert state + assert _events

This is the Testing Domain Logic in Isolation pattern. It verifies that business logic produces the correct state and raises the correct events. It does not verify that events reach their handlers or that projections update.

Use for: Business rules, invariants, state transitions, event payloads.

Level 2: Sync flow tests

Set event_processing = "sync" in the domain configuration. With sync processing, event handlers and projectors fire inline during the Unit of Work commit, in the same thread, in the same process, with no async machinery.

domain.process(command)
  -> handler fires synchronously
    -> aggregate persisted
      -> events dispatched inline
        -> event handlers fire inline
          -> projections updated

After domain.process() returns, the full chain has already executed. You can assert on projection state immediately:

# fragment
domain.process(PlaceOrder(...))

# No sleep, no engine, no async -- everything already ran
view = domain.view_for(OrderDashboard)
dashboard = view.get("ord-123")
assert dashboard.status == "placed"

Use for: Verifying the full command-to-projection chain, handler wiring, event routing, projection correctness. This is your primary integration testing tool.

Level 3: Async E2E tests

Use Engine(domain, test_mode=True) for tests that exercise the actual async processing infrastructure. The engine's run() method in test mode starts every subscription, lets them process messages until none of them finds more work, then performs a graceful shutdown. Messages that handlers raise along the way are processed too. No time.sleep() guesswork.

Test mode stops waiting after about a second if some subscription keeps finding work, for example while it retries a failing handler. A partitioned stream subscription cannot report that it is idle, so with one present test mode always waits the full second.

Test mode does not wait for a broker to redeliver a failed message. A broker subscription waits retry_delay_seconds before it nacks the message, and the inline broker then holds it back for its own retry_delay. Test mode stops before the redelivery. To test a handler that fails and then succeeds on retry, set both delays to 0:

[server.broker_subscription]
retry_delay_seconds = 0

[brokers.default]
provider = "inline"
retry_delay = 0

# fragment
engine = Engine(domain, test_mode=True)
engine.run()

# Engine has processed all pending messages
view = domain.view_for(OrderDashboard)
dashboard = view.get("ord-123")
assert dashboard.status == "placed"

Use for: Subscription configuration, priority lane routing, error handling paths (handle_error overrides), broker integration, outbox processing. Reserve Level 3 for tests that cannot be verified with sync processing.

When to use which level

Concern	Level 1	Level 2	Level 3
Business logic correctness	Yes	—	—
Event payloads and structure	Yes	—	—
Command-to-handler wiring	—	Yes	—
Event-to-handler routing	—	Yes	—
Projection updates	—	Yes	—
Multi-step event chains	—	Yes	Yes
Subscription configuration	—	—	Yes
Priority lane routing	—	—	Yes
Error handling paths	—	—	Yes
Outbox processing	—	—	Yes

Level 2 covers most integration needs. It runs at unit-test speed because there is no event loop, no network, and no async coordination. Reserve Level 3 for the small number of tests that truly need the async Engine.

---

Applying the Pattern

The examples below use a consistent e-commerce domain: an Order aggregate that raises OrderPlaced, an InventoryEventHandler that reserves stock, and an OrderDashboard projection that tracks order status.

Domain setup

The domain is in ecommerce/domain.py. InventoryEventHandler belongs to Inventory but listens to the order stream, so it names that stream with stream_category. Without it, the handler would listen to the inventory stream and never receive OrderPlaced.

from protean import Domain, handle
from protean.fields import Auto, Float, Identifier, Integer, String
from protean.utils.globals import current_domain

domain = Domain(name="ecommerce")


@domain.aggregate
class Order:
    order_id = Auto(identifier=True)
    customer_id = Identifier(required=True)
    product_id = Identifier(required=True)
    quantity = Integer(default=1)
    status = String(default="draft")
    total = Float(default=0.0)

    def place(self):
        self.status = "placed"
        self.raise_(
            OrderPlaced(
                order_id=self.order_id,
                customer_id=self.customer_id,
                product_id=self.product_id,
                quantity=self.quantity,
                total=self.total,
            )
        )


@domain.event(part_of=Order)
class OrderPlaced:
    order_id = Identifier(required=True)
    customer_id = Identifier(required=True)
    product_id = Identifier(required=True)
    quantity = Integer(required=True)
    total = Float(required=True)


@domain.command(part_of=Order)
class PlaceOrder:
    order_id = Identifier(required=True)
    customer_id = Identifier(required=True)
    product_id = Identifier(required=True)
    quantity = Integer(default=1)
    total = Float(required=True)


@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place_order(self, command: PlaceOrder):
        repo = current_domain.repository_for(Order)
        order = Order(
            order_id=command.order_id,
            customer_id=command.customer_id,
            product_id=command.product_id,
            quantity=command.quantity,
            total=command.total,
        )
        order.place()
        repo.add(order)


@domain.aggregate
class Inventory:
    product_id = Identifier(identifier=True)
    available_quantity = Integer(default=0)
    reserved_quantity = Integer(default=0)

    def reserve(self, order_id, quantity):
        self.available_quantity -= quantity
        self.reserved_quantity += quantity
        self.raise_(
            InventoryReserved(
                product_id=self.product_id, order_id=order_id, quantity=quantity
            )
        )


@domain.event(part_of=Inventory)
class InventoryReserved:
    product_id = Identifier(required=True)
    order_id = Identifier(required=True)
    quantity = Integer(required=True)


@domain.event_handler(part_of=Inventory, stream_category="ecommerce::order")
class InventoryEventHandler:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        repo = current_domain.repository_for(Inventory)
        inventory = repo.get(event.product_id)
        inventory.reserve(order_id=event.order_id, quantity=event.quantity)
        repo.add(inventory)


@domain.projection
class OrderDashboard:
    order_id = Identifier(identifier=True)
    customer_id = Identifier()
    status = String()
    total = Float()


@domain.projector(projector_for=OrderDashboard, aggregates=[Order])
class OrderDashboardProjector:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        repo = current_domain.repository_for(OrderDashboard)
        repo.add(
            OrderDashboard(
                order_id=event.order_id,
                customer_id=event.customer_id,
                status="placed",
                total=event.total,
            )
        )

Test fixtures

A tests/conftest.py that initializes the domain with sync processing and provides a per-test context. DomainFixture resets the data after each test:

# fragment
import pytest
from ecommerce.domain import domain

from protean.integrations.pytest import DomainFixture


@pytest.fixture(scope="session")
def app_fixture():
    domain.config["event_processing"] = "sync"
    domain.config["command_processing"] = "sync"

    fixture = DomainFixture(domain)
    fixture.setup()
    yield fixture
    fixture.teardown()


@pytest.fixture(autouse=True)
def _ctx(app_fixture):
    with app_fixture.domain_context():
        yield

Level 1: Domain unit tests

Test the aggregate's business logic and event raising directly:

# fragment
from ecommerce.domain import Order, OrderPlaced


class TestOrderPlacement:
    """Level 1: Test aggregate behavior in isolation."""

    def test_placing_order_transitions_status(self):
        order = Order(
            customer_id="cust-123",
            product_id="prod-1",
            total=99.95,
        )

        order.place()

        assert order.status == "placed"

    def test_placing_order_raises_order_placed_event(self):
        order = Order(
            customer_id="cust-123",
            product_id="prod-1",
            total=99.95,
        )

        order.place()

        assert len(order._events) == 1
        event = order._events[0]
        assert isinstance(event, OrderPlaced)
        assert event.order_id == order.order_id
        assert event.customer_id == "cust-123"
        assert event.total == 99.95

These tests run in microseconds. They verify that Order.place() produces the correct state and the correct event payload.

Level 2: Sync flow tests

Test the full command-to-projection chain with sync processing:

# fragment
from ecommerce.domain import Inventory, OrderDashboard, PlaceOrder, domain


class TestOrderFlowSync:
    """Level 2: Test the full chain with sync event processing."""

    def test_place_order_updates_projection(self):
        """Command -> handler -> aggregate -> event -> projector -> projection."""
        domain.repository_for(Inventory).add(
            Inventory(product_id="prod-1", available_quantity=100)
        )

        domain.process(
            PlaceOrder(
                order_id="ord-001",
                customer_id="cust-123",
                product_id="prod-1",
                total=99.95,
            )
        )

        # Projection is updated synchronously -- no sleep, no engine
        view = domain.view_for(OrderDashboard)
        dashboard = view.get("ord-001")

        assert dashboard.status == "placed"
        assert dashboard.customer_id == "cust-123"
        assert dashboard.total == 99.95

    def test_event_chain_fires_downstream_handlers(self):
        """Verify that OrderPlaced triggers InventoryEventHandler."""
        repo = domain.repository_for(Inventory)
        repo.add(Inventory(product_id="prod-1", available_quantity=100))

        domain.process(
            PlaceOrder(
                order_id="ord-003",
                customer_id="cust-123",
                product_id="prod-1",
                quantity=2,
                total=50.0,
            )
        )

        # In sync mode, the InventoryEventHandler fires inline
        # during the same UoW commit that persists the Order
        inventory = repo.get("prod-1")
        assert inventory.available_quantity == 98
        assert inventory.reserved_quantity == 2

These tests verify the wiring: commands reach handlers, events reach projectors, and projections reflect the correct state. They run at near unit-test speed because sync processing has no async overhead.

Level 3: Async E2E tests

Test the async engine for scenarios that require the full infrastructure. The async_events fixture switches event processing to async for one test and back to sync afterwards:

# fragment
import pytest
from ecommerce.domain import Inventory, Order, OrderDashboard, PlaceOrder, domain

from protean.exceptions import ObjectNotFoundError
from protean.server.engine import Engine


@pytest.fixture
def async_events():
    """Process events asynchronously for the length of one test."""
    domain.config["event_processing"] = "async"
    yield
    domain.config["event_processing"] = "sync"


@pytest.mark.usefixtures("async_events")
class TestOrderFlowAsync:
    """Level 3: Test with the real async Engine in test mode."""

    def test_full_async_flow(self):
        """Command -> handler -> event -> engine processes -> projection updated."""
        domain.repository_for(Inventory).add(
            Inventory(product_id="prod-1", available_quantity=10)
        )

        # Submit the command (sync command processing, async event processing)
        domain.process(
            PlaceOrder(
                order_id="ord-async-001",
                customer_id="cust-123",
                product_id="prod-1",
                total=75.00,
            )
        )

        # Nothing has processed the event yet
        view = domain.view_for(OrderDashboard)
        with pytest.raises(ObjectNotFoundError):
            view.get("ord-async-001")

        # Run the engine in test mode until every subscription is idle
        engine = Engine(domain, test_mode=True)
        engine.run()

        # After engine.run() completes, all pending events have been processed
        dashboard = view.get("ord-async-001")

        assert dashboard.status == "placed"
        assert dashboard.total == 75.00

        # Also verify the aggregate was persisted through the chain
        order = domain.repository_for(Order).get("ord-async-001")
        assert order.status == "placed"

Testing error handling paths

Level 3 is essential for testing handle_error() overrides. A decorator registers a handler on its domain when the module is imported, and pytest imports every test module before it runs any test. So a handler that fails on purpose goes on a small domain of its own, built in its test module. The shared domain never sees it:

# fragment
from typing import ClassVar

import pytest

from protean import Domain, handle
from protean.fields import Identifier
from protean.integrations.pytest import DomainFixture
from protean.server.engine import Engine

failures = Domain(name="failures", config={"event_processing": "async"})


@failures.aggregate
class Shipment:
    order_id = Identifier(required=True)

    def dispatch(self):
        self.raise_(ShipmentDispatched(order_id=self.order_id))


@failures.event(part_of=Shipment)
class ShipmentDispatched:
    order_id = Identifier(required=True)


@failures.event_handler(part_of=Shipment)
class FailingShipmentHandler:
    """A handler that deliberately fails, for testing error classification."""

    errors_captured: ClassVar[list[dict]] = []

    @handle(ShipmentDispatched)
    def on_dispatched(self, event: ShipmentDispatched):
        raise ConnectionError("Database temporarily unavailable")

    @classmethod
    def handle_error(cls, exc: Exception, message) -> None:
        cls.errors_captured.append(
            {
                "exception": exc,
                "message_type": message.metadata.headers.type,
            }
        )


@pytest.fixture
def failures_ctx():
    fixture = DomainFixture(failures)
    fixture.setup()
    FailingShipmentHandler.errors_captured.clear()
    with fixture.domain_context():
        yield
    fixture.teardown()


@pytest.mark.usefixtures("failures_ctx")
class TestErrorHandling:
    """Level 3: Verify error classification and handle_error behavior."""

    def test_handle_error_receives_exception(self):
        """Verify that handle_error is called when a handler raises."""
        shipment = Shipment(order_id="ord-err-001")
        shipment.dispatch()
        failures.repository_for(Shipment).add(shipment)

        engine = Engine(failures, test_mode=True)
        engine.run()

        # The handler failed, but handle_error captured the exception
        assert len(FailingShipmentHandler.errors_captured) > 0
        captured = FailingShipmentHandler.errors_captured[0]
        assert isinstance(captured["exception"], ConnectionError)
        assert "temporarily unavailable" in str(captured["exception"])
        assert captured["message_type"] == "Failures.ShipmentDispatched.v1"

Testing priority lane routing

Priority lanes route events through the outbox to broker streams, so the priority_lanes fixture switches the server to stream subscriptions and turns lanes on. The test processes the command with Priority.LOW and checks that the event went to the backfill stream and not the primary one:

# fragment
from protean.utils.processing import Priority


@pytest.fixture
def priority_lanes(async_events):
    """Route events through the outbox to broker streams, with priority lanes."""
    server = domain.config["server"]
    saved = dict(server)
    server["default_subscription_type"] = "stream"
    server["priority_lanes"] = {"enabled": True, "threshold": Priority.NORMAL}
    yield
    server.clear()
    server.update(saved)


@pytest.mark.usefixtures("priority_lanes")
class TestPriorityLanes:
    """Level 3: Verify priority-based event routing."""

    def test_low_priority_events_route_to_backfill_lane(self):
        """Events below the threshold are routed to the backfill stream."""
        domain.repository_for(Inventory).add(
            Inventory(product_id="prod-1", available_quantity=10)
        )

        domain.process(
            PlaceOrder(
                order_id="ord-pri-001",
                customer_id="cust-456",
                product_id="prod-1",
                total=50.0,
            ),
            priority=Priority.LOW,
        )

        # In test mode, backfill events are still processed (the primary
        # stream is empty, so the backfill stream is polled).
        engine = Engine(domain, test_mode=True)
        engine.run()

        broker = domain.brokers["default"]
        backfill = broker.read("ecommerce::order:backfill", "check", 10)
        primary = broker.read("ecommerce::order", "check", 10)
        assert [message["data"]["order_id"] for _, message in backfill] == [
            "ord-pri-001"
        ]
        assert primary == []

        dashboard = domain.view_for(OrderDashboard).get("ord-pri-001")
        assert dashboard.status == "placed"

---

Anti-Patterns

Using time.sleep() for synchronization

# fragment
# WRONG: Arbitrary sleep durations are fragile.
def test_order_flow(self):
    domain.process(PlaceOrder(...))
    time.sleep(5)  # "Works on my machine" -- fails in CI
    dashboard = domain.repository_for(OrderDashboard).get("ord-123")
    assert dashboard.status == "placed"

Why it's wrong: The sleep duration is a guess. Too short and tests are flaky. Too long and the suite is slow.

Fix: Use Level 2 (sync processing) or Level 3 (Engine(test_mode=True)) for deterministic execution.

Testing handlers without the full chain

# fragment
# WRONG: Testing a handler with a manually constructed event.
def test_inventory_handler(self):
    event = OrderPlaced(
        order_id="ord-123",
        customer_id="cust-456",
        total=100.0,
    )
    handler = InventoryEventHandler()
    handler.on_order_placed(event)

    # Passes, but in production the event might have different
    # metadata, come from a different stream, or not arrive at all.

Why it's wrong: The test misses the wiring. A misconfigured part_of or stream category means the handler is never invoked in production, but this test would still pass.

Fix: Use Level 2 sync flow tests to verify the full chain.

Skipping E2E tests entirely

# fragment
# WRONG: "Our unit tests cover everything."
# No Level 2 or Level 3 tests exist. The team discovers in production
# that the OrderDashboardProjector was never registered, or that
# events are routed to the wrong stream category.

Why it's wrong: Unit tests cannot verify that components are wired together correctly. Registration, stream routing, and projection mapping only surface when the full chain runs.

Fix: Add Level 2 sync flow tests for every command-to-projection path.

Mixing async and sync in the same test

# fragment
# WRONG: Sending a command synchronously but expecting async event processing.
def test_confused_flow(self):
    domain.config["event_processing"] = "async"
    domain.process(PlaceOrder(...))

    # Events were dispatched to the event store, but no engine is running.
    # This assertion will fail because no one is processing events.
    dashboard = domain.view_for(OrderDashboard).get("ord-123")
    assert dashboard is not None  # Fails!

Why it's wrong: With async event processing, events are written to the event store but not processed until an engine reads them. Without starting the engine, the events sit unprocessed.

Fix: Either use sync processing (Level 2) or start the engine in test mode (Level 3) after dispatching the command.

---

Summary

Aspect	Level 1: Domain unit	Level 2: Sync flow	Level 3: Async E2E
What it tests	Business logic, events	Full chain wiring	Async infrastructure
Processing mode	None (direct calls)	event_processing = "sync"	Engine(test_mode=True)
Speed	Microseconds	Milliseconds	Seconds
Infrastructure	None	None	Event store, broker
Deterministic	Yes	Yes	Yes (runs until every subscription is idle, for about one second at most)
Catches	Logic bugs, wrong events	Wiring, routing, projections	Subscription config, priority lanes, error paths
Test count	Many (majority)	Some (key flows)	Few (infra-specific)
Typical assertion	order.status, order._events	domain.view_for(P).get(id)	domain.view_for(P).get(id) after engine.run()

Most event-driven flow testing belongs at Level 2. Set event_processing = "sync", call domain.process(), and assert on projection state immediately. The sync pipeline exercises the same handlers and projectors as production, without async complexity. Reserve Level 3 for subscription configuration, priority lanes, and error handling paths.

---

Related reading

Patterns:

• Testing Domain Logic in Isolation: Level 1: unit testing domain elements.
• Idempotent Event Handlers: Handlers safe for replay and re-testing.
• Dual-Mode Testing: Run tests against in-memory and real adapters.

Guides:

• Application Tests: Testing command handlers and services.
• Integration Tests: Full integration testing.
• Event Sourcing Tests: Fluent test DSL for ES aggregates.