Coverage for faststream / nats / helpers / state.py: 87%

1from typing import Protocol, TypeVar (empty)

2 

3from nats.aio.client import Client (empty)

4from nats.js import JetStreamContext (empty)

5 

6from faststream.exceptions import IncorrectState (empty)

7 

8ClientT = TypeVar("ClientT", Client, JetStreamContext) (empty)

9 

10 

11class ConnectionState(Protocol[ClientT]): (empty)

12 connection: ClientT (empty)

13 

14 

15class EmptyConnectionState(ConnectionState[ClientT]): (empty)

16 __slots__ = () (empty)

17 

18 @property (empty)

19 def connection(self) -> ClientT: (empty)

20 raise IncorrectState 

21 

22 @connection.setter (empty)

23 def connection(self, v: ClientT) -> None: (empty)

24 raise IncorrectState 

25 

26 

27class ConnectedState(ConnectionState[ClientT]): (empty)

28 __slots__ = ("connection",) (empty)

29 

30 def __init__(self, connection: ClientT) -> None: (empty)

31 self.connection: ClientT = connection (empty)