Testing¶
test_harness runs the mediators the application already uses and records what they did. It does not add a bus. Commands and queries stay cqrs.Request. Import it from cqrs.testing. It is not in cqrs.__all__.
Two entries return the same harness. Building a stand takes the same mapper and middleware arguments as bootstrap. The builder passes compensation_retry_delay=0 unless the test passes a delay. around does not change the delay of a mediator that is already built.
from cqrs.testing import test_harness
async with test_harness(
container,
commands_mapper=commands_mapper,
queries_mapper=queries_mapper,
domain_events_mapper=domain_events_mapper,
events_mapper=events_mapper,
streaming_mapper=streaming_mapper,
sagas_mapper=sagas_mapper,
middlewares=middlewares,
scope_strategy=cqrs.ScopeStrategy.NONE,
) as harness:
await harness.mediator.send(CreateOrder(sku="book"))
around wraps mediators the application already built. For the block it wraps resolve, the broker, saga storage, and the listener, and puts them back in finally.
async with test_harness.around(
mediator,
event_mediator=event_mediator,
streaming_mediator=streaming_mediator,
saga_mediator=saga_mediator,
outbox=outbox,
inbox=inbox,
) as harness:
await mediator.send(CreateOrder(sku="book"))
test_harness(...) creates a memory outbox and a memory inbox and stores them on harness.outbox_repository and harness.inbox_repository. It does not bind those ports. The test binds the port.
TestContainer is register(type, instance) and async resolve. For ScopeStrategy.SEND, pass di.Container, not TestContainer.
If the mediator holds DevnullMessageBroker, the harness replaces it with CapturingMessageBroker for the block and restores it afterwards. Do not read DevnullMessageBroker.MESSAGE_BUS.
contains is a boolean. assert_contains, assert_one, and assert_empty raise AssertionError and list what was recorded. There is no pytest plugin. The same calls work under unittest.
An invocation record has handler, method (handle, act, or compensate), argument, argument_type, and handler_instance.
Journals¶
Three delivery channels stay separate. A test that asks published for a domain event gets a miss.
invocationsrecords entry intohandle,act, andcompensate: the class, the method, and the argument. A chain step, a fallback, a saga step, and anEventMediatorhandler are here.EventMediatorgoes throughEventDispatcher, notEventEmitter, so the emitter listener does not see it.handledis aDomainEventthatEventEmittergave to handlers after a command, a stream, or a saga step.publishedis a message that reachedMessageBroker.send_message. That is a directNotificationEvent, and laterEventProducer.outboxis a row the handler stored. It is not inpublisheduntilproduce_available.requestsis eachsend: the request type, and the response or the exception. A command and a query are bothcqrs.Request.yieldsis what a stream yielded, in order.middlewaresrecords which middleware saw the request and whether it calledhandle.handler_errorsis a domain-handler error thatsendswallows and logs.unhandledis a domain event whose map is empty.sagasis status, context, and step history from storage.inboxis insert, a duplicate, and processing throughInboxProcessor.
assert harness.published.contains(OrderCreated) is false for a domain event. assert_contains says the type is on handled or invocations and did not go to the broker.
Handlers¶
-
Command
Assert the command on the requests and invocations journals.
-
Query
After a read, the handled, published, and outbox journals are empty.
-
Domain event
Assert the domain event on the handled and invocations journals.
-
Notification
Assert the notice on the published journal, including its topic.
-
Parallel handlers
Assert both handler classes on the invocations journal.
-
Handler errors
The command result returns from send; the error is on the handler_errors journal.
-
Unhandled event
Assert the event type on the unhandled journal.
-
Event mediator
Assert the handler on the invocations journal, method handle.
-
Request fallback
Assert both classes on invocations; the requests journal holds the fallback response.
-
Event fallback
Assert both classes on the invocations journal.
-
Circuit breaker
An open fake ICircuitBreaker leaves the primary off the invocations journal.
-
Middleware
The middlewares journal records called_next; requests holds the middleware response.
-
Scope SEND
Assert invocation order on the invocations journal under ScopeStrategy.SEND.
-
Scope HANDLER and NONE
Assert both handlers on the invocations journal; order is not checked.
-
Chain
Assert the entered step on the invocations journal.
-
Streaming
Assert values on the yields journal and the event on handled.
-
Streaming fallback
Assert both handlers on the yields and invocations journals.
-
Saga step
Assert act on the invocations journal and the running saga count.
-
Saga compensation
Assert compensate on the invocations journal and the failed saga status.
-
Saga fallback
Assert both classes on the invocations journal; the saga stays completed.
-
Saga circuit breaker
A fake ISagaStepCircuitBreaker leaves the primary off the invocations journal.
-
Saga recovery
Recover through the sagas journal; a finished step does not act again.
-
Inbox
Assert the command on the requests journal and the processed inbox row.
-
Inbox failure
Assert NOT_PROCESSED and flush_counter on the inbox journal.
-
Outbox
Assert the row on the outbox journal, then the notice on published.
-
Outbox broker error
A broker error leaves the outbox row not produced; the published journal stays empty.
Models¶
-
Dataclass
The response is on requests. The harness adds no validator.
-
Pydantic
ValidationError is raised by the constructor, before send. A valid send returns the response.
-
JSON
Default path for a published message. content_type is None.
-
Protobuf
Published message content_type and payload_bytes. The harness does not import protobuf.
-
di
ScopeStrategy.SEND on di.Container shares one session. Pass di.Container, not TestContainer.
Recipes¶
-
FastAPI
TestClient and dependency_overrides. Assert requests and outbox. The harness does not import FastAPI.
-
FastStream
TestKafkaBroker delivers. The harness checks invocations or the inbox and does not import FastStream.
-
SQLAlchemy
Session rollback and SqlAlchemyOutboxedEventRepository on SQLite. The harness does not import SQLAlchemy.
-
SQLModel
SqlModelOutboxedEventRepository under the sqlmodel marker. The harness does not import SQLModel.
-
dishka
around wraps a mediator that is already built. The harness does not import dishka.
-
dependency-injector
around wraps the mediator. The adapter does not implement SupportsScope, so scope_strategy is not SEND.
-
Kafka and AMQP
The unit test uses CapturingMessageBroker. Real Kafka and AMQP adapters stay in integration tests.
Outside the harness¶
These pages are here so the feature is not dropped. They are not a test_harness run.
-
Mermaid
CoRMermaid and SagaMermaid draw sequence and class diagrams. They take no part in send.
-
Compressor
ZlibCompressor is checked by the serializer tests. The harness sees payload_bytes as finished bytes.
-
Alembic
rebind is a migration, not a handler run. The fence is the short form.
-
Transaction log tailing
A process that reads the database log. It is not mediator.send and is not run through the harness.