Bases: TestBroker[RabbitBroker]
A class to test RabbitMQ brokers.
Source code in faststream/testing/broker.py
| def __init__(
self,
broker: Broker,
with_real: bool = False,
connect_only: Optional[bool] = None,
) -> None:
self.with_real = with_real
self.broker = broker
if connect_only is None:
try:
connect_only = is_contains_context_name(
self.__class__.__name__,
TestApp.__name__,
)
except Exception: # pragma: no cover
warnings.warn(
(
"\nError `{e!r}` occurred at `{self.__class__.__name__}` AST parsing."
"\n`connect_only` is set to `False` by default."
),
category=RuntimeWarning,
stacklevel=1,
)
connect_only = False
self.connect_only = connect_only
self._fake_subscribers: List[SubscriberProto[Any]] = []
|
with_real instance-attribute
broker instance-attribute
connect_only instance-attribute
create_publisher_fake_subscriber staticmethod
create_publisher_fake_subscriber(broker, publisher)
Source code in faststream/rabbit/testing.py
| @staticmethod
def create_publisher_fake_subscriber(
broker: RabbitBroker,
publisher: AsyncAPIPublisher,
) -> Tuple["LogicSubscriber", bool]:
sub: Optional[LogicSubscriber] = None
for handler in broker._subscribers.values():
if _is_handler_suitable(
handler,
publisher.routing,
{},
publisher.exchange,
):
sub = handler
break
if sub is None:
is_real = False
sub = broker.subscriber(
queue=publisher.routing,
exchange=publisher.exchange,
)
else:
is_real = True
return sub, is_real
|