Skip to content

Kafka and AMQP

  • Back to Testing Overview

    Return to the Testing overview page with all topics.

    Back to Overview

The harness does not import Kafka or AMQP. The unit test uses CapturingMessageBroker and reads published. The real Kafka and AMQP adapters stay in integration tests, so those clients stay out of published.

See Message brokers and Event producing.

from cqrs.testing import CapturingMessageBroker

broker = CapturingMessageBroker()
mediator = bootstrap.bootstrap(container, message_broker=broker, commands_mapper=commands_mapper)
async with test_harness.around(mediator, broker=broker) as harness:
    await mediator.send(CreateOrder(sku="book"))
    harness.published.assert_one(OrderCreatedNotice, topic="order_events")

KafkaMessageBroker construction stays in the integration tests:

broker = cqrs.message_brokers.kafka.KafkaMessageBroker(producer)
mediator = bootstrap.bootstrap(container, message_broker=broker, commands_mapper=commands_mapper)