Skip to content

Integration Tests

DDD CQRS ES

Integration tests verify that your application works correctly with real infrastructure, databases, message brokers, event stores, and caches. They exercise the same code paths as production, giving you confidence that your domain logic persists and communicates correctly through real adapters.

Like application tests, we recommend using pytest-bdd for integration tests. The BDD structure works just as well here. The only difference is the domain is configured with real infrastructure instead of in-memory adapters.

Integration tests are written the same way as application tests, with feature files and step definitions, but they swap in real adapters (PostgreSQL, Redis, etc.) via configuration overrides. Use pytest markers to keep them separate from your fast in-memory suite.

Configuring for Real Infrastructure

The key difference from application tests is the domain configuration. Override your database, broker, or event store settings in a dedicated conftest.py and use DomainFixture to manage the lifecycle:

# fragment
# tests/integration/conftest.py
import pytest
from pytest_bdd import given

from protean.integrations.pytest import DomainFixture

from myapp import domain


@pytest.fixture(scope="session")
def app_fixture():
    domain.config["databases"]["default"] = {
        "provider": "postgresql",
        "database_uri": "postgresql://postgres:postgres@localhost:5432/myapp_test",
    }
    domain.config["brokers"]["default"] = {
        "provider": "redis",
        "URI": "redis://localhost:6379/0",
    }
    domain.config["event_processing"] = "sync"
    domain.config["command_processing"] = "sync"

    fixture = DomainFixture(domain)
    fixture.setup()    # domain.init() + create schema
    yield fixture
    fixture.teardown()  # drop schema


@pytest.fixture(autouse=True)
def _ctx(app_fixture):
    with app_fixture.domain_context():  # resets all data on exit
        yield


@given("the domain is initialized")
def domain_initialized():
    pass

Note

DomainFixture.setup() calls domain.init() and creates database tables for all configured providers. teardown() drops them. domain_context() resets all data (providers, brokers, event store) after each test. The same application code and domain elements are used, only the infrastructure changes.

Using Server Test Mode

For integration tests that exercise async message flows (event handlers, projectors, process managers), use the server's test mode. It processes all available messages and cascading events, then exits.

The examples in this section test the domain below. Placing an order raises OrderPlaced. An event handler reserves stock and raises InventoryReserved, and a second handler marks the order shipped. The fixtures set both processing modes to "async", so these handlers run only when the engine runs:

The domain and fixtures under test 
# fragment
from protean import Domain, handle
from protean.fields import HasMany, Identifier, Integer, String
from protean.utils.globals import current_domain

domain = Domain(name="Warehouse")


@domain.event(part_of="Order")
class OrderPlaced:
    order_id: Identifier(required=True)
    product_id: String(required=True)
    quantity: Integer(required=True)


@domain.event(part_of="Order")
class OrderCancelled:
    order_id: Identifier(required=True)


@domain.entity(part_of="Order")
class OrderItem:
    product_id: String(max_length=50, required=True)
    quantity: Integer(min_value=1, required=True)


@domain.aggregate
class Order:
    customer_id: String(max_length=50)
    status: String(max_length=20, default="placed")
    items = HasMany(OrderItem)

    @classmethod
    def create(cls, customer_id, items, **kwargs):
        order = cls(customer_id=customer_id, items=items, **kwargs)
        for item in items:
            order.raise_(
                OrderPlaced(
                    order_id=order.id,
                    product_id=item.product_id,
                    quantity=item.quantity,
                )
            )
        return order


@domain.event(part_of="Reservation")
class InventoryReserved:
    order_id: Identifier(required=True)


@domain.aggregate
class Reservation:
    order_id: Identifier(identifier=True)
    product_id: String(max_length=50, required=True)
    quantity: Integer(required=True)

    @classmethod
    def reserve(cls, order_id, product_id, quantity):
        reservation = cls(order_id=order_id, product_id=product_id, quantity=quantity)
        reservation.raise_(InventoryReserved(order_id=order_id))
        return reservation


@domain.repository(part_of=Reservation)
class ReservationRepository:
    def get_by_order(self, order_id):
        return self._dao.query.filter(order_id=order_id).all().first


@domain.command(part_of=Order)
class PlaceOrder:
    order_id: Identifier(identifier=True)
    product_id: String(required=True)
    qty: Integer(required=True)


@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place_order(self, command: PlaceOrder):
        order = Order.create(
            customer_id="anonymous",
            items=[OrderItem(product_id=command.product_id, quantity=command.qty)],
            id=command.order_id,
        )
        current_domain.repository_for(Order).add(order)


@domain.event_handler(part_of=Order)
class InventoryHandler:
    """Reserves stock for every order placed."""

    @handle(OrderPlaced)
    def reserve(self, event: OrderPlaced):
        reservation = Reservation.reserve(
            event.order_id, event.product_id, event.quantity
        )
        current_domain.repository_for(Reservation).add(reservation)


@domain.event_handler(part_of=Reservation)
class ShippingHandler:
    """Ships an order once its stock is reserved."""

    @handle(InventoryReserved)
    def ship(self, event: InventoryReserved):
        repo = current_domain.repository_for(Order)
        order = repo.get(event.order_id)
        order.status = "shipped"
        repo.add(order)

# fragment
import pytest
from fulfillment import domain

from protean.integrations.pytest import DomainFixture


@pytest.fixture(scope="session")
def app_fixture():
    # Handlers run asynchronously, in the server engine.
    domain.config["event_processing"] = "async"
    domain.config["command_processing"] = "async"

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


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

The tests read an order back with a small helper in the test module:

# fragment
def get_order(order_id):
    return domain.repository_for(Order).get(order_id)

# fragment
from protean.server import Engine


def test_order_creates_inventory_reservation():
    """Test that creating an order reserves inventory."""
    # Arrange: Create order (raises events)
    with domain.domain_context():
        order = Order.create(
            customer_id="123", items=[OrderItem(product_id="ABC", quantity=5)]
        )
        domain.repository_for(Order).add(order)

    # Act: Process events in test mode
    engine = Engine(domain, test_mode=True)
    engine.run()

    # Assert: Verify inventory was reserved
    with domain.domain_context():
        reservation = domain.repository_for(Reservation).get_by_order(order.id)
        assert reservation is not None
        assert reservation.quantity == 5

Test mode handles cascading events automatically. If processing one event raises another, the engine continues until all chains are complete:

# fragment
def test_order_fulfillment_flow():
    """Test complete order fulfillment flow."""
    # Order created -> Inventory reserved -> Order shipped
    with domain.domain_context():
        order = Order.create(
            customer_id="123", items=[OrderItem(product_id="ABC", quantity=5)]
        )
        domain.repository_for(Order).add(order)

    # Process all cascading events
    engine = Engine(domain, test_mode=True)
    engine.run()

    with domain.domain_context():
        order = domain.repository_for(Order).get(order.id)
        assert order.status == "shipped"

You can also invoke test mode from the CLI:

protean server --domain=my_domain --test-mode

See Run the Server for CLI details.

Driving flows with process_and_wait

The Engine(domain, test_mode=True).run() calls above are the low-level primitive. For most tests, protean.testing provides higher-level helpers. With these helpers, the same test body works in both synchronous and asynchronous processing modes. The helpers also hand you the events and errors, so you do not read outbox rows or event-store streams yourself.

process_and_wait(command, domain) processes a command through its full runtime path (handler → aggregate → UoW commit → outbox → engine → subscription → projector) and waits for the effects to settle. In sync mode the chain runs inline; in async mode it drains a bounded test-mode engine before returning. It returns a ProcessResult:

# fragment
from protean.testing import process_and_wait


def test_placing_an_order_reserves_inventory():
    outcome = process_and_wait(
        PlaceOrder(order_id="o1", product_id="ABC", qty=5), domain
    )

    assert outcome.succeeded
    assert OrderPlaced in outcome.events
    assert outcome.events[OrderPlaced].order_id == "o1"

    with domain.domain_context():
        reservation = domain.repository_for(Reservation).get("o1")
        assert reservation.quantity == 5

A ProcessResult exposes exactly what an integration test asserts on:

  • result: The handler's return value (sync) or the enqueued command's store position (async).
  • events: An EventLog of every event raised in the command's correlation chain.
  • error / succeeded / failed: the synchronous or submission-time error (an unregistered command, an expired deadline, a duplicate key, an enrichment ValidationError). Asynchronous handler failures happen after the command is enqueued, are absorbed by the engine (retries or DLQ), and are not surfaced here.

Asserting on events with EventLog

outcome.events is an EventLog with Pythonic access by type or index:

# fragment
def test_the_event_log():
    outcome = process_and_wait(
        PlaceOrder(order_id="o1", product_id="ABC", qty=5), domain
    )

    assert OrderPlaced in outcome.events  # membership by type
    assert outcome.events[OrderPlaced].order_id == "o1"  # first event of a type
    assert outcome.events.get(OrderCancelled) is None  # safe access, no KeyError
    assert outcome.events.types == [OrderPlaced, InventoryReserved]
    assert len(outcome.events) == 2

Waiting for a condition with drain

When you drive the flow yourself, or need to wait for a later cascade to settle, drain runs the test-mode engine until a predicate is satisfied. It replaces the copy-pasted for _ in range(N): Engine(...).run() loop:

# fragment
from protean.testing import drain


def test_order_ships_after_reservation():
    domain.process(PlaceOrder(order_id="o1", product_id="ABC", qty=5))

    drain(domain, until=lambda: get_order("o1").status == "shipped")

    assert get_order("o1").status == "shipped"

drain is bounded by max_cycles (default 5) so a never-satisfied until cannot hang the test. If the budget is exhausted before the condition settles, drain returns and emits a UserWarning, so an unmet expectation is never silently swallowed. An engine pass stops once nothing is left to process and takes at most about a second, so raise max_cycles only for flows that genuinely need more passes.

Full-Flow Feature Files

Integration tests exercise complete flows from command to projection, verifying that every layer works together through real infrastructure.

Command Through to Projection

# tests/integration/features/order_flow.feature
@integration
Feature: End-to-End Order Flow
    Verify the complete order lifecycle from book creation
    through order placement, including event-driven side effects.

    Background:
        Given the domain is initialized

    Scenario: Full order lifecycle
        Given a book "Dune" is added to the catalog at 15.99
        And the book has 10 copies in stock
        And a customer "Alice" exists
        When "Alice" places an order for 2 copies of "Dune"
        Then the order status should be "CONFIRMED"
        And "Dune" should have 8 copies in stock
        And the catalog projection should show 2 copies sold for "Dune"

    Scenario: Order updates customer history across aggregates
        Given a book "1984" is added to the catalog at 11.99
        And the book has 5 copies in stock
        And a customer "Bob" exists
        When "Bob" places an order for 2 copies of "1984"
        Then "Bob" should have 1 order in their history

Persistence Round-Trip

# tests/integration/features/persistence.feature
@integration
Feature: Persistence
    Verify that aggregates survive a round-trip through
    the real database.

    Background:
        Given the domain is initialized

    Scenario: Book survives persistence round-trip
        Given a book "The Great Gatsby" by "F. Scott Fitzgerald" at 12.99
        When the book is persisted and reloaded
        Then the reloaded book should have title "The Great Gatsby"
        And the reloaded book should have price 12.99

    Scenario: Updated aggregate is persisted correctly
        Given a book "Dune" by "Frank Herbert" at 15.99
        And the book is persisted
        When the book price is updated to 19.99
        And the book is persisted and reloaded
        Then the reloaded book should have price 19.99

Step Definitions for Integration Tests

Step definitions are written the same way as in application tests. They import your domain and use domain.process() and domain.repository_for():

# fragment
# tests/integration/test_order_flow.py
from myapp import domain
from myapp.commands import AddBook, PlaceOrder
from myapp.models import BookCatalog, Customer, Inventory, Order
from pytest_bdd import given, parsers, scenarios, then, when

scenarios("features/order_flow.feature")


@given(
    parsers.parse('a book "{title}" is added to the catalog at {price:f}'),
    target_fixture="book_id",
)
def add_book(title, price):
    return domain.process(
        AddBook(title=title, author="Test Author", price_amount=price)
    )


@given(
    parsers.parse("the book has {quantity:d} copies in stock"),
)
def stock_book(book_id, quantity):
    inventory = Inventory(book_id=book_id, title="", quantity=quantity)
    domain.repository_for(Inventory).add(inventory)


@given(
    parsers.parse('a customer "{name}" exists'),
    target_fixture="customer",
)
def existing_customer(name):
    customer = Customer(name=name)
    domain.repository_for(Customer).add(customer)
    return customer


@when(
    parsers.parse('"{name}" places an order for {qty:d} copies of "{title}"'),
    target_fixture="order",
)
def place_order(customer, book_id, name, qty, title):
    domain.process(
        PlaceOrder(
            customer_id=customer.id,
            items=[{"book_id": book_id, "quantity": qty}],
        )
    )
    orders = domain.repository_for(Order).query.all()
    return orders.items[0]


@then(
    parsers.parse(
        'the catalog projection should show {sold:d} copies sold for "{title}"'
    )
)
def check_projection(book_id, sold, title):
    catalog = domain.repository_for(BookCatalog).get(book_id)
    assert catalog.copies_sold == sold


@then(parsers.parse('the order status should be "{status}"'))
def check_status(order, status):
    assert domain.repository_for(Order).get(order.id).status == status


@then(parsers.parse('"{title}" should have {expected:d} copies in stock'))
def check_stock(book_id, title, expected):
    inventory = domain.repository_for(Inventory).find_by_book(book_id)
    assert inventory.quantity == expected


@then(parsers.parse('"{name}" should have {count:d} order in their history'))
def check_history(customer, name, count):
    customer = domain.repository_for(Customer).get(customer.id)
    assert len(customer.order_history) == count

The persistence feature's steps go in tests/integration/test_persistence.py:

# fragment
# tests/integration/test_persistence.py
from myapp import domain
from myapp.models import Book, Money
from pytest_bdd import given, parsers, scenarios, then, when

scenarios("features/persistence.feature")


@given(
    parsers.parse('a book "{title}" by "{author}" at {price:f}'),
    target_fixture="book",
)
def new_book(title, author, price):
    return Book(title=title, author=author, price=Money(amount=price))


@given("the book is persisted")
def persist_book(book):
    domain.repository_for(Book).add(book)


@when(parsers.parse("the book price is updated to {price:f}"))
def update_price(book, price):
    book.price = Money(amount=price)


@when("the book is persisted and reloaded", target_fixture="reloaded")
def persist_and_reload(book):
    domain.repository_for(Book).add(book)
    return domain.repository_for(Book).get(book.id)


@then(parsers.parse('the reloaded book should have title "{title}"'))
def check_title(reloaded, title):
    assert reloaded.title == title


@then(parsers.parse("the reloaded book should have price {price:f}"))
def check_price(reloaded, price):
    assert reloaded.price.amount == price

Separating Integration Tests

Use Gherkin tags and pytest markers to separate integration tests from fast in-memory tests:

@integration
Feature: Persistence
    ...

Run them selectively:

# Fast local development — skip integration tests
pytest tests/ -m "not integration"

# CI pipeline — run everything
pytest tests/

# Run only integration tests
pytest tests/ -m integration

You can also separate them by directory structure:

tests/
├── conftest.py                  # Shared root fixtures
├── unit/                        # Domain model tests (always fast)
│   └── ...
├── bdd/                         # Application tests (in-memory)
│   └── ...
└── integration/                 # Integration tests (real infra)
    ├── conftest.py              # Real adapter configuration
    ├── features/
    │   ├── order_flow.feature
    │   └── persistence.feature
    ├── test_order_flow.py
    └── test_persistence.py

Testing with Different Adapters

Prefer --protean-env for adapter switching

Instead of writing per-directory conftest.py overrides, consider using Protean's built-in dual-mode testing support. Define a [memory] overlay in domain.toml and switch modes from the command line:

pytest --protean-env memory   # In-memory — fast, no Docker
pytest                        # Real adapters — default (test) overlay

No test code or fixture changes needed. See the Dual-Mode Testing pattern for the complete setup.

For projects that need finer-grained control or want to switch individual adapters independently, use environment variables to switch configuration. This fixture goes in tests/integration/conftest.py, and domain is your application's domain (from myapp import domain):

# fragment
# tests/integration/conftest.py
import os

import pytest
from myapp import domain

from protean.integrations.pytest import DomainFixture


@pytest.fixture(scope="session")
def app_fixture():
    db_provider = os.environ.get("TEST_DB", "memory")

    if db_provider == "postgresql":
        domain.config["databases"]["default"] = {
            "provider": "postgresql",
            "database_uri": os.environ["DATABASE_URL"],
        }
    elif db_provider == "sqlite":
        domain.config["databases"]["default"] = {
            "provider": "sqlite",
            "database_uri": "sqlite:///test.db",
        }
    # Default: in-memory (no override needed)

    domain.config["event_processing"] = "sync"
    domain.config["command_processing"] = "sync"

    fixture = DomainFixture(domain)
    fixture.setup()  # calls domain.init() + creates schema
    yield fixture
    fixture.teardown()  # drops schema


@pytest.fixture(autouse=True)
def _ctx(app_fixture):
    with app_fixture.domain_context():  # resets all data on exit
        yield

Then switch adapters from the command line:

# Default: in-memory
pytest tests/integration/

# With PostgreSQL
TEST_DB=postgresql DATABASE_URL=postgresql://localhost/myapp_test pytest tests/integration/

# With SQLite
TEST_DB=sqlite pytest tests/integration/

The integration feature files and step definitions then run unchanged against whichever adapter TEST_DB names. The domain logic is identical, only the infrastructure changes.

Per-Test Data Cleanup

DomainFixture.domain_context() automatically resets data in all providers, brokers, and the event store after each test. If you use the recommended _ctx fixture pattern above, no additional cleanup fixture is needed.

Coverage Reporting

Run your full test suite with coverage to verify you're hitting the 100% target on business logic:

# Run all tests with coverage
pytest --cov=myapp --cov-report=html tests/

# Run with a minimum coverage threshold
pytest --cov=myapp --cov-fail-under=100 tests/

Exclude setup files from coverage in your pyproject.toml:

[tool.coverage.run]
source = ["myapp"]
omit = ["myapp/__init__.py", "myapp/config.py"]