Testing

Testing handlers directly, faking the mediator, async tests, and the global-registry isolation gotcha.

PyMediate's framework independence is what makes testing straightforward: handlers are plain callables with injected dependencies, so most of your tests never need a mediator, a web framework, or a database at all.

Testing handlers directly

The fastest and most common test: construct the handler with fakes, call it, assert on the response. No mediator required.

from dataclasses import dataclass
from pymediate import Request, Handler

@dataclass
class CreateUserResponse:
    user_id: int
    username: str

@dataclass
class CreateUserRequest(Request[CreateUserResponse]):
    username: str
    email: str

class CreateUserHandler(Handler[CreateUserRequest]):
    def __init__(self, database):
        self.database = database

    def __call__(self, request: CreateUserRequest) -> CreateUserResponse:
        user_id = self.database.create_user(request.username, request.email)
        return CreateUserResponse(user_id=user_id, username=request.username)

def test_create_user_handler():
    handler = CreateUserHandler(database=FakeDatabase())

    response = handler(CreateUserRequest(username="alice", email="alice@example.com"))

    assert response.username == "alice"
    assert response.user_id > 0

Prefer a fake or in-memory implementation over a mock where practical — it exercises more real behavior for about the same amount of test code.

Testing through the mediator

Reserve this for the handful of tests that specifically verify request routing, pipeline behavior wiring, or handler composition — not as a replacement for direct handler tests:

from pymediate import Mediator, Services

def test_create_user_through_mediator():
    services = Services()
    services.add(CreateUserHandler(database=FakeDatabase()))
    mediator = Mediator(services.provider())

    response = mediator.send(CreateUserRequest(username="alice", email="alice@example.com"))

    assert response.username == "alice"

Faking the mediator

When testing a handler that itself dispatches through an injected mediator (see handler composition), fake the mediator rather than wiring up every handler it might call:

class FakeMediator:
    def __init__(self):
        self.sent = []

    def send(self, request):
        self.sent.append(request)
        return FakeResponseFor(request)

def test_place_order_sends_expected_requests():
    mediator = FakeMediator()
    handler = PlaceOrderHandler(mediator=mediator, database=FakeDatabase())

    handler(PlaceOrderRequest(...))

    assert isinstance(mediator.sent[0], ChargePaymentRequest)
    assert isinstance(mediator.sent[1], SendEmailRequest)

Testing async handlers

Async handlers are async def __call__ methods — test them with pytest.mark.asyncio like any other coroutine:

import pytest
from pymediate import Services
from pymediate.aio import Mediator

@pytest.mark.asyncio
async def test_async_handler_directly():
    handler = FetchDataHandler(http_client=FakeHttpClient())

    response = await handler(FetchDataRequest(url="https://api.example.com/data"))

    assert response.data is not None

@pytest.mark.asyncio
async def test_async_handler_through_mediator():
    services = Services()
    services.add(FetchDataHandler(http_client=FakeHttpClient()))
    mediator = Mediator(services.provider())

    response = await mediator.send(FetchDataRequest(url="https://api.example.com/data"))

    assert response.data is not None

Services is imported from pymediate, not pymediate.aio, even in async tests — only Handler and Mediator have separate sync/async variants.

A test-isolation gotcha: the handler registry is global

Each Handler[RequestT] subclass registers itself against its request type the moment the class body executes — not per Services instance, but in a single registry shared by the whole process. Defining a second Handler for a request type that already has one raises HandlerAlreadyRegisteredError — and that applies across two test files just as much as in application code:

# test_a.py
class TempHandler(Handler[SharedRequest]):
    def __call__(self, request: SharedRequest) -> SharedResponse:
        return SharedResponse(value=1)

# test_b.py — if SharedRequest is the same class, this collides at import/collection time
class TempHandler(Handler[SharedRequest]):
    def __call__(self, request: SharedRequest) -> SharedResponse:
        return SharedResponse(value=2)
# HandlerAlreadyRegisteredError: Handler already registered for 'SharedRequest'

Vary behavior through the constructor instead of redefining the class:

class ConfigurableHandler(Handler[SharedRequest]):
    def __init__(self, value: int):
        self.value = value

    def __call__(self, request: SharedRequest) -> SharedResponse:
        return SharedResponse(value=self.value)

# test_a.py
handler = ConfigurableHandler(value=1)
# test_b.py
handler = ConfigurableHandler(value=2)

If you genuinely need two distinct handler implementations, give them distinct request types — see troubleshooting for the full set of options.

Fixtures and organization

Ordinary pytest fixtures work well for the pieces you construct repeatedly:

import pytest

@pytest.fixture
def fake_database():
    return FakeDatabase()

@pytest.fixture
def create_user_handler(fake_database):
    return CreateUserHandler(database=fake_database)

def test_create_user(create_user_handler):
    response = create_user_handler(
        CreateUserRequest(username="alice", email="alice@example.com")
    )
    assert response.user_id > 0

Organize by concern: keep handler tests next to the handlers they cover, and keep the smaller set of mediator/pipeline integration tests separate.

Next steps

Error handling — domain vs. framework errors and where to map them
Type safety — what mypy catches vs. what's validated at runtime
Troubleshooting — common registration and configuration mistakes