Coverage for faststream / redis / configs / broker.py: 93%

1from dataclasses import dataclass (empty)

2from typing import TYPE_CHECKING (empty)

3 

4from faststream._internal.configs import BrokerConfig (empty)

5from faststream.exceptions import IncorrectState (empty)

6 

7if TYPE_CHECKING: (empty)

8 from faststream.redis.parser import MessageFormat 

9 from faststream.redis.publisher.producer import RedisFastProducer 

10 

11 from .state import ConnectionState 

12 

13 

14@dataclass(kw_only=True) (empty)

15class RedisBrokerConfig(BrokerConfig): (empty)

16 producer: "RedisFastProducer" (empty)

17 connection: "ConnectionState" (empty)

18 

19 message_format: type["MessageFormat"] (empty)

20 

21 async def connect(self) -> None: (empty)

22 self.producer.connect(self.fd_config._serializer) (empty)

23 await self.connection.connect() (empty)

24 

25 async def disconnect(self) -> None: (empty)

26 await self.connection.disconnect() (empty)

27 

28 

29@dataclass(kw_only=True) (empty)

30class RedisRouterConfig(BrokerConfig): (empty)

31 @property (empty)

32 def connection(self) -> ConnectionError: (empty)

33 raise IncorrectState