Skip to content

TestNatsBroker

faststream.nats.TestNatsBroker #

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

Bases: TestBroker[NatsBroker]

Source code in faststream/broker/test.py 
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:
    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 staticmethod #

create_publisher_fake_subscriber(
    broker: NatsBroker, publisher: Publisher
) -> HandlerCallWrapper[Any, Any, Any]
Source code in faststream/nats/test.py 
24
25
26
27
28
29
30
31
32
33
@staticmethod
def create_publisher_fake_subscriber(
    broker: NatsBroker,
    publisher: Publisher,
) -> HandlerCallWrapper[Any, Any, Any]:
    @broker.subscriber(publisher.subject, _raw=True)
    def f(msg: Any) -> None:
        pass

    return f

patch_publisher staticmethod #

patch_publisher(broker: NatsBroker, publisher: Any) -> None
Source code in faststream/nats/test.py 
20
21
22
@staticmethod
def patch_publisher(broker: NatsBroker, publisher: Any) -> None:
    publisher._producer = broker._producer

remove_publisher_fake_subscriber staticmethod #

remove_publisher_fake_subscriber(
    broker: NatsBroker, publisher: Publisher
) -> None
Source code in faststream/nats/test.py 
39
40
41
42
43
@staticmethod
def remove_publisher_fake_subscriber(
    broker: NatsBroker, publisher: Publisher
) -> None:
    broker.handlers.pop(Handler.get_routing_hash(publisher.subject), None)