Testing

Most handler tests can construct the handler and call it directly. Tests through a mediator are useful when the subject is routing, service wiring, or pipeline behavior order.

Test handler behavior directly

A handler is a callable with ordinary constructor dependencies. The following module defines the domain types once, uses a fake dependency, and tests the handler without a mediator:

from dataclasses import dataclass
from typing import Protocol

import pytest

from pymediate import Mediator, Request, RequestHandler, Services


@dataclass(frozen=True)
class OrderReceipt:
    order_id: int
    summary: str


@dataclass(frozen=True)
class PlaceOrder(Request[OrderReceipt]):
    customer_id: int
    item: str
    quantity: int


class OrderStore(Protocol):
    async def add(self, item: str, quantity: int) -> int: ...


class FakeOrderStore:
    def __init__(self) -> None:
        self.saved: list[tuple[str, int]] = []

    async def add(self, item: str, quantity: int) -> int:
        self.saved.append((item, quantity))
        return 42


class PlaceOrderHandler(RequestHandler[PlaceOrder]):
    def __init__(self, store: OrderStore) -> None:
        self._store = store

    async def __call__(self, request: PlaceOrder) -> OrderReceipt:
        order_id = await self._store.add(request.item, request.quantity)
        return OrderReceipt(
            order_id=order_id,
            summary=f"{request.quantity} × {request.item}",
        )


@pytest.mark.asyncio
async def test_place_order_handler() -> None:
    store = FakeOrderStore()
    handler = PlaceOrderHandler(store)

    receipt = await handler(PlaceOrder(customer_id=7, item="tea", quantity=2))

    assert receipt == OrderReceipt(order_id=42, summary="2 × tea")
    assert store.saved == [("tea", 2)]


@pytest.mark.asyncio
async def test_place_order_dispatch() -> None:
    store = FakeOrderStore()
    services = Services(PlaceOrderHandler(store))
    mediator = Mediator(services)

    receipt = await mediator.send(
        PlaceOrder(customer_id=7, item="tea", quantity=2),
    )

    assert receipt.order_id == 42

The first test isolates handler behavior. The second also verifies that the handler instance is available through the service provider and that send() routes the exact request type to it.

Prefer a small fake or in-memory implementation when its behavior matters to the test. A mock is also suitable when the test only needs to assert a specific interaction. If a handler dispatches other requests, inject a narrow application protocol for that operation and fake the protocol rather than constructing the application's entire mediator graph.

Test each dispatch form at its boundary

Use the smallest boundary that covers the behavior under test:

For notification publishing, assert against injected fakes instead of handler completion order. Async notification handlers run concurrently. For streams, consume only as many chunks as the test needs when laziness is part of the contract.

Keep definitions stable across tests

Request, stream, and notification handler classes register when Python defines the class. The registries are process-wide, not attached to a Services instance.

This has two consequences for a test suite:

• Defining a second request or stream handler class for the same request type raises HandlerAlreadyRegisteredError, even when the definitions are in different test files.
• Notification subscribers accumulate. Publishing a notification attempts to resolve every subscriber class registered for that exact notification type. A subscriber instance omitted from the test's service provider causes ServiceNotFoundError before any subscriber runs.

Define each test request, notification, and handler class once in a module or fixture support module. Vary behavior through constructor dependencies. Use a distinct request or notification type when a test genuinely needs a different handler definition. The registry-reset functions under pymediate._internal are test infrastructure for PyMediate itself, not public application API.

Build a new Services collection for tests that need isolated instances. A collection is immutable once constructed, so tests never leak services into one another. To reuse a production wiring with one service swapped, combine it with | — the right operand wins:

from pymediate import Mediator, Services

# The application's own wiring, imported rather than duplicated in the test.
from myapp.wiring import build_services

fake = FakeOrderStore()
mediator = Mediator(build_services() | Services(PlaceOrderHandler(fake)))

receipt = await mediator.send(PlaceOrder(customer_id=7, item="tea", quantity=2))
assert fake.saved == [("tea", 2)]

Test the API you import

Async tests await request and notification dispatch and use async for for streams. Sync tests call the corresponding methods directly and use for for streams.

The namespaces intentionally define six different async and sync objects: RequestHandler, NotificationHandler, StreamRequestHandler, Mediator, PipelineBehavior, and Next. Shared types such as Request, StreamRequest, Notification, Services, ServiceProvider, and the exception classes are identical objects in both namespaces.

Continue

• Type safety separates static checks from definition-time validation.
• Troubleshooting maps common exceptions to configuration checks.
• Error handling distinguishes application errors from PyMediate errors.