Bases: TestBroker[RedisBroker]
A class to test Redis 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/redis/testing.py
| @staticmethod
def create_publisher_fake_subscriber(
broker: RedisBroker,
publisher: "AsyncAPIPublisher",
) -> Tuple["LogicSubscriber", bool]:
sub: Optional[LogicSubscriber] = None
named_property = publisher.subscriber_property(name_only=True)
visitors = (ChannelVisitor(), ListVisitor(), StreamVisitor())
for handler in broker._subscribers.values(): # pragma: no branch
for visitor in visitors:
if visitor.visit(**named_property, sub=handler):
sub = handler
break
if sub is None:
is_real = False
sub = broker.subscriber(**publisher.subscriber_property(name_only=False))
else:
is_real = True
return sub, is_real
|