Coverage for faststream / nats / broker / state.py: 81%

1from typing import TYPE_CHECKING (empty)

2 

3from faststream.exceptions import IncorrectState (empty)

4 

5if TYPE_CHECKING: (empty)

6 from nats.aio.client import Client 

7 from nats.js import JetStreamContext 

8 

9 

10class BrokerState: (empty)

11 def __init__(self) -> None: (empty)

12 self._connected = False (empty)

13 

14 self._stream: JetStreamContext | None = None (empty)

15 self._connection: Client | None = None (empty)

16 

17 @property (empty)

18 def connection(self) -> "Client": (empty)

19 if not self._connection: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true(empty)

20 msg = "Connection is not available yet. Please, connect the broker first." 

21 raise IncorrectState(msg) 

22 return self._connection (empty)

23 

24 @property (empty)

25 def stream(self) -> "JetStreamContext": (empty)

26 if not self._stream: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true(empty)

27 msg = "Stream is not available yet. Please, connect the broker first." 

28 raise IncorrectState(msg) 

29 return self._stream (empty)

30 

31 def __bool__(self) -> bool: (empty)

32 return self._connected (empty)

33 

34 def connect(self, connection: "Client", stream: "JetStreamContext") -> None: (empty)

35 self._connection = connection (empty)

36 self._stream = stream (empty)

37 self._connected = True (empty)

38 

39 def disconnect(self) -> None: (empty)

40 self._connection = None (empty)

41 self._stream = None (empty)

42 self._connected = False (empty)