Coverage for faststream / mqtt / subscriber / specification.py: 90%

1from typing import TYPE_CHECKING, Any (empty)

2 

3from faststream._internal.endpoint.subscriber import SubscriberSpecification (empty)

4from faststream.mqtt.broker.config import MQTTBrokerConfig (empty)

5from faststream.specification.asyncapi.utils import resolve_payloads (empty)

6from faststream.specification.schema import Message, Operation, SubscriberSpec (empty)

7from faststream.specification.schema.bindings import ( (empty)

8 ChannelBinding, 

9 OperationBinding, 

10 mqtt as mqtt_bindings, 

11) 

12 

13from .config import MQTTSubscriberSpecificationConfig (empty)

14 

15if TYPE_CHECKING: (empty)

16 from faststream._internal.endpoint.subscriber.call_item import CallsCollection 

17 

18 

19class MQTTSubscriberSpecification( (empty)

20 SubscriberSpecification[MQTTBrokerConfig, MQTTSubscriberSpecificationConfig], 

21): 

22 def __init__( (empty)

23 self, 

24 _outer_config: "MQTTBrokerConfig", 

25 specification_config: "MQTTSubscriberSpecificationConfig", 

26 calls: "CallsCollection[Any]", 

27 ) -> None: 

28 super().__init__(_outer_config, specification_config, calls) (empty)

29 

30 @property (empty)

31 def topic(self) -> str: (empty)

32 base = f"{self._outer_config.prefix}{self.config.topic}" (empty)

33 if self.config.shared: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true(empty)

34 return f"$share/{self.config.shared}/{base}" 

35 return base (empty)

36 

37 @property (empty)

38 def name(self) -> str: (empty)

39 if self.config.title_: (empty)

40 return self.config.title_ (empty)

41 return f"{self.topic}:{self.call_name}" (empty)

42 

43 def get_schema(self) -> dict[str, SubscriberSpec]: (empty)

44 payloads = self.get_payloads() (empty)

45 

46 return { (empty)

47 self.name: SubscriberSpec( 

48 description=self.description, 

49 operation=Operation( 

50 message=Message( 

51 title=f"{self.name}:Message", 

52 payload=resolve_payloads(payloads), 

53 ), 

54 bindings=OperationBinding( 

55 mqtt=mqtt_bindings.OperationBinding( 

56 qos=self.config.qos, 

57 ), 

58 ), 

59 ), 

60 bindings=ChannelBinding( 

61 mqtt=mqtt_bindings.ChannelBinding( 

62 topic=self.topic, 

63 qos=self.config.qos, 

64 ), 

65 ), 

66 ), 

67 }