Coverage for faststream / mqtt / broker / config.py: 86%

1from dataclasses import dataclass, field (empty)

2from typing import TYPE_CHECKING, Any, Literal, Optional, cast (empty)

3 

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

5from faststream.exceptions import FeatureNotSupportedException, IncorrectState (empty)

6from faststream.mqtt.publisher.producer import ZmqttFakeProducer (empty)

7from faststream.opentelemetry.middleware import TelemetryMiddleware (empty)

8 

9if TYPE_CHECKING: (empty)

10 import zmqtt 

11 

12 from faststream._internal.types import BrokerMiddleware 

13 from faststream.mqtt.publisher.producer import ZmqttBaseProducer 

14 

15 

16MQTTVersionUnset = cast("str", object()) (empty)

17 

18 

19@dataclass(kw_only=True) (empty)

20class MQTTBrokerConfig(BrokerConfig): (empty)

21 version: Literal["3.1.1", "5.0", "unset"] = "unset" (empty)

22 

23 producer: "ZmqttBaseProducer" = field(default_factory=ZmqttFakeProducer) (empty)

24 _client: Optional["zmqtt.MQTTClient"] = field(default=None, init=False, repr=False) (empty)

25 

26 def __post_init__(self) -> None: (empty)

27 for m in self.broker_middlewares: (empty)

28 self._validate_middleware(m) (empty)

29 

30 @property (empty)

31 def client(self) -> "zmqtt.MQTTClient": (empty)

32 if self._client is None: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true(empty)

33 msg = "MQTT broker is not connected. Call connect() first." 

34 raise IncorrectState(msg) 

35 return self._client (empty)

36 

37 def connect(self, client: "zmqtt.MQTTClient") -> None: (empty)

38 self._client = client (empty)

39 self.producer.connect(client, self.fd_config._serializer) (empty)

40 

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

42 self._client = None (empty)

43 self.producer.disconnect() (empty)

44 

45 def add_middleware(self, middleware: "BrokerMiddleware[Any]") -> None: (empty)

46 self._validate_middleware(middleware) (empty)

47 return super().add_middleware(middleware) (empty)

48 

49 def insert_middleware(self, middleware: "BrokerMiddleware[Any]") -> None: (empty)

50 self._validate_middleware(middleware) 

51 return super().insert_middleware(middleware) 

52 

53 def _validate_middleware(self, middleware: "BrokerMiddleware[Any]") -> None: (empty)

54 if self.version == "3.1.1" and isinstance(middleware, TelemetryMiddleware): (empty)

55 msg = "Opentelementry don`t work in 3.1.1 mqtt" (empty)

56 raise FeatureNotSupportedException(msg) (empty)