Testing Utilities¶

The framework provides testing utilities for building domain objects, inspecting events, and creating stub containers for integration testing.

Imports¶

from aod.testing import build, events_of, assert_event_emitted, assert_no_events, check_invariant
from aod.testing import FakeDomain
from aod.testing.doubles import (
    SpyLogger,
    SpyEventBus,
    port_stub,
    spy_session,
    spy_adapter_container,
    spy_command_handler,
    spy_query_handler,
    spy_async_command_handler,
    spy_async_query_handler,
)

Helper Functions¶

`build`¶

def build(cls: type[T], **kwargs: Any) -> T

Create an instance of a domain class skipping validation.

Parameters¶

Parameter	Type	Description
`cls`	`type[T]`	The class to instantiate. Must be an `Entity`, `RootEntity`, `ValueObject`, or any domain object subclass.
`**kwargs`	`Any`	Field values to set on the instance.

Returns¶

T — An instance of cls without validation.

user = build(User, id=1, name="Alice")

`events_of`¶

def events_of(obj: object) -> list[Event]

Extract all events emitted by a domain object via its _event_emitter.

Parameters¶

Parameter	Type	Description
`obj`	`object`	The domain object (`Entity`, `ValueObject`, `Service`, etc.).

Returns¶

list[Event] — All events emitted by the object since construction or last clear.

events = events_of(user)

`assert_event_emitted`¶

def assert_event_emitted(events: Sequence[Event], event_type: type[Event], **attrs: Any) -> Event

Assert that a specific event type was emitted with matching field values.

Parameters¶

Parameter	Type	Description
`events`	`Sequence[Event]`	The list of events to search (from `events_of()`).
`event_type`	`type[Event]`	The expected event class.
`**attrs`	`Any`	Field values that the event must match.

Returns¶

Event — The matching event instance.

events = events_of(user)
assert_event_emitted(events, UserCreatedEvent, user_id=1)

`assert_no_events`¶

def assert_no_events(events: Sequence[Event]) -> None

Assert that no events were emitted.

assert_no_events(events_of(service))

`check_invariant`¶

def check_invariant(cls: type, invariant_name: str, data: dict[str, Any] | None = None, **kwargs: Any) -> None

Run a single invariant validator against an instance created via build().

check_invariant(User, "valid_email", email="not-an-email")

Container Testing with `spy_adapter_container`¶

The recommended approach for testing use cases is to create a spy version of your container. This replaces sessions and ports with stubs that record calls and let you configure return values.

from aod.infrastructure import AdapterContainer
from aod.testing.doubles import spy_adapter_container

container = spy_adapter_container(
    AdapterContainer(sessions={MySession}, handlers=[CreateUserHandler, GetUserHandler])
)

# Replace a use case's run() with a fixed return or exception
container.stub_use_case(CreateUserUseCase, returns=user)
container.stub_use_case(GetUserUseCase, raises=ValueError("not found"))

# adapt() returns a fully wired instance
uc = container.adapt(CreateUserUseCase)
result = uc.run(user_id=1, name="Alice")  # → user

# Assert events and handler calls
assert result.name == "Alice"
assert container.get_handler(CreateUser).handle.call_count == 1

To control individual CommandPort/QueryPort fields inside a use case that has multiple handlers, configure handlers directly without stubbing the whole use case:

# Each handler's handle() is a MagicMock returning None by default
container.get_handler(CreateUser).handle.return_value = user
container.get_handler(GetUser).handle.side_effect = ValueError("db down")

uc = container.adapt(ManageUserUseCase)
uc.run(dto)  # create.handle() returns user, get.handle() raises

`get_session_stub`¶

Access the stub for a given session class. Each stub method records calls and lets you configure return values:

stub = container.get_session_stub(MySession)
stub.is_dirty.return_value = True      # always returns True
stub.is_dirty.side_effect = [True, False]  # first True, then False
stub.is_dirty.called                  # True if called at least once
stub.is_dirty.call_count              # number of calls
stub.is_dirty.call_args_list          # list of call objects, each with .args and .kwargs
stub.begin.called                     # tracks begin() too
stub.commit.called                    # commit is called by the UseCase wrapper

`get_port_stub`¶

Access the stub for a given port field name. Works with any Port subclass registered on the container:

container = spy_adapter_container(AdapterContainer(logger=SpyLogger()))
stub = container.get_port_stub("logger")
stub.info.return_value = None
stub.info.called
stub.info.call_args_list

`get_handler`¶

Retrieve the handler for a given contract. Handler methods are also stubbed:

handler = container.get_handler(CreateUser)
assert handler.handle.called

`get_handler_stub`¶

Access the handler stub for a given handler class. Works like get_port_stub for handlers:

stub = container.get_handler_stub(CreateUserHandler)
stub.handle.return_value = None
stub.handle.called

`stub_use_case` with `returns=` / `raises=`¶

Configure a use case stub before calling adapt. returns= stubs instance.run to return the given value; raises= makes it raise an exception:

container.stub_use_case(CreateUserUseCase, returns=42)
use_case = container.adapt(CreateUserUseCase)
result = use_case.run(user_id=1)  # returns 42

`stub_projection` with `read_returns` / `read_raises` / `write_returns` / `write_raises`¶

Configure a projection stub before calling adapt:

container.stub_projection(MyProjection, read_returns=[], write_returns=None)
proj = container.adapt(MyProjection)
proj.read(model)    # returns []
proj.write(model)   # returns None

`port_stub`¶

For testing ports outside a container context, create stubs directly:

from aod.testing.doubles import port_stub

StubLogger = port_stub(Logger)
logger = StubLogger()
logger.info("test")
assert logger.info.called

Every stub method is a unittest.mock.MagicMock or AsyncMock. Configure return values and inspect calls with the standard mock API:

logger.info("message", user_id=1)
assert logger.info.called
assert logger.info.call_count == 1
entry = logger.info.call_args_list[0]
entry.args    # ("message",)
entry.kwargs  # {"user_id": 1}

Stub Control¶

Every stub method is a unittest.mock mock object:

Property / Method	Description
`.return_value = value`	Always return this value
`.side_effect = exc`	Raise an exception
`.side_effect = [v1, v2]`	Return different values on successive calls
`.called`	Whether the method was called
`.call_count`	Number of calls
`.call_args_list`	List of `call` objects — each has `.args` and `.kwargs`

Spy Handler Factories¶

`spy_command_handler` / `spy_query_handler`¶

from aod.testing.doubles import (
    spy_command_handler,
    spy_query_handler,
    spy_async_command_handler,
    spy_async_query_handler,
)

Factory functions that create handler port spies. Each call returns a fresh instance with its own handle() mock — no shared state.

Parameters¶

Parameter	Type	Description
`returns`	`Any`	Value that `handle()` returns on every call. Default: `None`.
`raises`	`Exception \\| None`	Exception that `handle()` raises on every call. Takes precedence over `returns`.

Sync Handler Spies¶

spy = spy_command_handler(returns=user)
result = spy.handle(my_command)  # returns `user`
assert spy.handle.call_count == 1

spy = spy_query_handler(raises=ValueError("not found"))
spy.handle(my_query)  # raises ValueError

Async Handler Spies¶

spy = spy_async_command_handler(returns=user)
result = await spy.handle(my_command)  # returns `user` (awaitable)
assert spy.handle.call_count == 1

Use in UseCases¶

class CreateUserUseCase(UseCase):
    save_user: CommandPort[CreateUser]

    def run(self, name: str) -> User:
        return self.save_user.handle(CreateUser(name=name))

spy = spy_command_handler(returns=user)
uc = CreateUserUseCase(save_user=spy)
uc.run(name="Alice")
assert spy.handle.call_count == 1

Spy Classes¶

SpyLogger¶

from aod.testing.doubles import SpyLogger

SpyLogger is a ready-made implementation of Logger(Port):

logger = SpyLogger()
logger.info("order placed", order_id=1)
assert logger.info.called

SpyEventBus¶

from aod.testing.doubles import SpyEventBus

bus = SpyEventBus()
bus.publish(OrderPlaced(order_id=1))
assert bus.publish.called

`spy_session`¶

from aod.testing.doubles import spy_session

Create a stub class from any Session or AsyncSession subclass. Every method (including custom methods your concrete session defines) becomes a MagicMock (or AsyncMock for async sessions). is_dirty() returns False by default.

StubPg = spy_session(PgSession)
session = StubPg()
session.is_dirty.return_value = True
session.query.return_value = [{"id": 1}]
session.set.return_value = True  # custom PgSession method
assert session.begin.called
assert session.commit.called

In the spy container, container.get_session_stub(Session) already does this for you.

FakeDomain¶

from aod.testing import FakeDomain

FakeDomain(Generic[T]) generates test data for domain objects using polyfactory.

Constructor¶

FakeDomain(model_cls: type[T], **defaults: Any)

Parameter	Type	Description
`model_cls`	`type[T]`	The domain class to build. Must be `Entity`, `RootEntity`, or `ValueObject`.
`**defaults`	`Any`	Default values for specific fields.

`call`¶

def __call__(self, **overrides: Any) -> T

Build an instance with auto-generated fields for any not provided.

users = FakeDomain(User)
alice = users(name="Alice")  # auto-generates other fields

`batch`¶

def batch(self, count: int, overrides_list: list[dict[str, Any]] | None = None) -> list[T]

Build multiple instances.

batch = users.batch(5)  # 5 users with auto-generated fields

Common Testing Patterns¶

Testing Use Cases with Spy Container¶

from aod.testing.doubles import spy_adapter_container


container = spy_adapter_container(AdapterContainer(sessions={MySession}, handlers=[CreateUserHandler, GetUserHandler]))
use_case = container.adapt(CreateUserUseCase)
use_case.run(user_id=1, name="Alice")

assert_event_emitted(use_case.events, UserCreatedEvent, user_id=1)
assert container.get_handler(CreateUser).handle.called

Testing Entities¶

from aod.testing import build, events_of, assert_event_emitted

class User(RootEntity):
    id: int
    name: str

    def __post_init__(self) -> None:
        self._event_emitter.emit(UserCreatedEvent(user_id=self.id))

user = build(User, id=1, name="Alice")
events = events_of(user)
assert_event_emitted(events, UserCreatedEvent, user_id=1)

Testing Value Objects¶

from aod.testing import build

class Address(ValueObject):
    street: str
    city: str

addr = build(Address, street="123 Main", city="NYC")

Testing Utilities¶

Imports¶

Helper Functions¶

build¶

Parameters¶

Returns¶

events_of¶

Parameters¶

Returns¶

assert_event_emitted¶

Parameters¶

Returns¶

assert_no_events¶

check_invariant¶

Container Testing with spy_adapter_container¶

get_session_stub¶

get_port_stub¶

get_handler¶

get_handler_stub¶

stub_use_case with returns= / raises=¶

stub_projection with read_returns / read_raises / write_returns / write_raises¶

port_stub¶

Stub Control¶

Spy Handler Factories¶

spy_command_handler / spy_query_handler¶

Parameters¶

Sync Handler Spies¶

Async Handler Spies¶

Use in UseCases¶

Spy Classes¶

SpyLogger¶

SpyEventBus¶

spy_session¶

FakeDomain¶

Constructor¶

__call__¶

batch¶

Common Testing Patterns¶

Testing Use Cases with Spy Container¶

Testing Entities¶

Testing Value Objects¶

Next Steps¶

`build`¶

`events_of`¶

`assert_event_emitted`¶

`assert_no_events`¶

`check_invariant`¶

Container Testing with `spy_adapter_container`¶

`get_session_stub`¶

`get_port_stub`¶

`get_handler`¶

`get_handler_stub`¶

`stub_use_case` with `returns=` / `raises=`¶

`stub_projection` with `read_returns` / `read_raises` / `write_returns` / `write_raises`¶

`port_stub`¶

`spy_command_handler` / `spy_query_handler`¶

`spy_session`¶

`call`¶

`batch`¶