Skip to content

TestBroker

faststream.broker.test.TestBroker #

TestBroker(
    broker: Broker,
    with_real: bool = False,
    connect_only: Optional[bool] = None,
)

Bases: Generic[Broker]

A class to represent a test broker.

Initialize a class instance.

PARAMETER DESCRIPTION
broker

An instance of the Broker class.

TYPE: Broker

with_real

Whether to use a real broker.

TYPE: bool DEFAULT: False

connect_only

Whether to only connect to the broker.

TYPE: Optional[bool] DEFAULT: None

Source code in faststream/broker/test.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def __init__(
    self,
    broker: Broker,
    with_real: bool = False,
    connect_only: Optional[bool] = None,
) -> None:
    """Initialize a class instance.

    Args:
        broker: An instance of the Broker class.
        with_real: Whether to use a real broker.
        connect_only: Whether to only connect to the broker.

    """
    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 as e:  # pragma: no cover
            # TODO: remove with 0.5.0
            warnings.warn(
                (
                    f"\nError `{e!r}` occurred at `{self.__class__.__name__}` AST parsing"
                    "\nPlease, report us by creating an Issue with your TestClient use case"
                    "\nhttps://github.com/airtai/faststream/issues/new?labels=bug&template=bug_report.md&title=Bug:%20TestClient%20AST%20parsing"
                ),
                category=RuntimeWarning,
                stacklevel=1,
            )

            connect_only = False

    self.connect_only = connect_only

broker instance-attribute #

broker = broker

connect_only instance-attribute #

connect_only = connect_only

with_real instance-attribute #

with_real = with_real

create_publisher_fake_subscriber abstractmethod staticmethod #

create_publisher_fake_subscriber(
    broker: Broker, publisher: Any
) -> HandlerCallWrapper[Any, Any, Any]
Source code in faststream/broker/test.py 
239
240
241
242
243
244
@staticmethod
@abstractmethod
def create_publisher_fake_subscriber(
    broker: Broker, publisher: Any
) -> HandlerCallWrapper[Any, Any, Any]:
    raise NotImplementedError()

patch_publisher abstractmethod staticmethod #

patch_publisher(broker: Broker, publisher: Any) -> None
Source code in faststream/broker/test.py 
256
257
258
259
@staticmethod
@abstractmethod
def patch_publisher(broker: Broker, publisher: Any) -> None:
    raise NotImplementedError()

remove_publisher_fake_subscriber abstractmethod staticmethod #

remove_publisher_fake_subscriber(
    broker: Broker, publisher: Any
) -> None
Source code in faststream/broker/test.py 
246
247
248
249
@staticmethod
@abstractmethod
def remove_publisher_fake_subscriber(broker: Broker, publisher: Any) -> None:
    raise NotImplementedError()