Skip to content

TestBroker

faststream.broker.test.TestBroker #

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

Bases: Generic[Broker]

Source code in faststream/broker/test.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 as e:
            warnings.warn(
                (
                    f"\nError `{repr(e)}` occured at `{self.__class__.__name__}` AST parsing"
                    "\nPlease, report us by creating an Issue with your TestClient usecase"
                    "\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

@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

@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

@staticmethod
@abstractmethod
def remove_publisher_fake_subscriber(broker: Broker, publisher: Any) -> None:
    raise NotImplementedError()