aio_microservice.amqp.testing

Attributes

ServiceT

Classes

AmqpBroker
TestAmqpBroker	A class to test RabbitMQ brokers.

Module Contents

aio_microservice.amqp.testing.ServiceT

class aio_microservice.amqp.testing.AmqpBroker(broker)

Parameters:

broker (faststream.rabbit.RabbitBroker)

__getattr__(attr)

Parameters:

attr (str)

Return type:

Any

__setattr__(attr, val)

Implement setattr(self, name, value).

Parameters:

• attr (str)

• val (Any)

Return type:

None

get_published_messages(queue=None, exchange=None)

Parameters:

• queue (str | None)

• exchange (str | None)

Return type:

list[Any]

reset_published_messages()

Return type:

None

class aio_microservice.amqp.testing.TestAmqpBroker(service, with_real=False)

Bases: faststream.rabbit.TestRabbitBroker

A class to test RabbitMQ brokers.

Parameters:

• service (ServiceT)

• with_real (bool)

async __aenter__()

Return type:

AmqpBroker