Skip to content

TestNatsBroker

faststream.nats.testing.TestNatsBroker #

TestNatsBroker(
    broker: NatsBroker,
    /,
    *,
    with_real: bool = False,
    connect_only: bool | None = None,
)

TestNatsBroker(
    *brokers: NatsBroker,
    with_real: bool = False,
    connect_only: bool | None = None,
)

TestNatsBroker(
    *brokers: NatsBroker,
    with_real: bool = False,
    connect_only: bool | None = None,
)

Bases: TestBroker[NatsBroker, EnterType]

A class to test NATS brokers.

Source code in faststream/nats/testing.py 
77
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    *brokers: NatsBroker,
    with_real: bool = False,
    connect_only: bool | None = None,
) -> None:
    super().__init__(
        *brokers,
        with_real=with_real,
        connect_only=connect_only,
    )

with_real instance-attribute #

with_real = with_real

brokers instance-attribute #

brokers = brokers

connect_only instance-attribute #

connect_only = connect_only

create_publisher_fake_subscriber #

create_publisher_fake_subscriber(
    broker: NatsBroker, publisher: LogicPublisher
) -> tuple[LogicSubscriber[Any], bool]
Source code in faststream/nats/testing.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def create_publisher_fake_subscriber(
    self,
    broker: NatsBroker,
    publisher: "LogicPublisher",
) -> tuple["LogicSubscriber[Any]", bool]:
    publisher_stream = publisher.stream.name if publisher.stream else None

    sub: LogicSubscriber[Any] | None = None
    for handler in (s for b in self.brokers for s in b.subscribers):
        handler = cast("LogicSubscriber[Any]", handler)
        if _is_handler_matches(handler, publisher.subject.template, publisher_stream):
            sub = handler
            break

    if sub is None:
        is_real = False
        sub = broker.subscriber(
            publisher.subject.template,
            persistent=False,
            stream=publisher_stream,
        )
    else:
        is_real = True

    return sub, is_real