Coverage for faststream / mqtt / parser.py: 91%

1from contextlib import suppress (empty)

2from typing import TYPE_CHECKING, Any (empty)

3 

4import zmqtt (empty)

5 

6from faststream._internal._compat import json_loads (empty)

7from faststream.message import StreamMessage, decode_message (empty)

8 

9from .message import MQTTMessage (empty)

10 

11if TYPE_CHECKING: (empty)

12 from faststream._internal.basic_types import DecodedMessage 

13 

14 

15class MQTTBaseParser: (empty)

16 """Base parser for MQTT messages — shared parse + decode logic.""" 

17 

18 async def parse_message(self, msg: zmqtt.Message) -> MQTTMessage: (empty)

19 raise NotImplementedError 

20 

21 async def decode_message(self, msg: "StreamMessage[Any]") -> "DecodedMessage": (empty)

22 return decode_message(msg) (empty)

23 

24 

25class MQTTParserV311(MQTTBaseParser): (empty)

26 """Parser for MQTT 3.1.1 messages — raw payload, no metadata.""" 

27 

28 async def parse_message(self, msg: zmqtt.Message) -> MQTTMessage: (empty)

29 return MQTTMessage( (empty)

30 raw_message=msg, 

31 body=msg.payload, 

32 headers={}, 

33 content_type=None, 

34 reply_to="", 

35 correlation_id=None, 

36 ) 

37 

38 async def decode_message(self, msg: "StreamMessage[Any]") -> "DecodedMessage": (empty)

39 body: bytes = msg.body (empty)

40 with suppress(Exception): (empty)

41 m: DecodedMessage = json_loads(body) (empty)

42 return m (empty)

43 with suppress(UnicodeDecodeError): (empty)

44 return body.decode() (empty)

45 return body 

46 

47 

48class MQTTParserV5(MQTTBaseParser): (empty)

49 """Parser for MQTT 5.0 messages. (empty)

50 

51 Extracts content_type, response_topic, correlation_data, and 

52 user_properties from PUBLISH properties when available. 

53 """ 

54 

55 async def parse_message(self, msg: zmqtt.Message) -> MQTTMessage: (empty)

56 props = msg.properties (empty)

57 content_type: str | None = None (empty)

58 reply_to: str = "" (empty)

59 correlation_id: str | None = None (empty)

60 headers: dict[str, Any] = {} (empty)

61 

62 if props is not None: 62 ↛ 69line 62 didn't jump to line 69 because the condition on line 62 was always true(empty)

63 content_type = props.content_type (empty)

64 reply_to = props.response_topic or "" (empty)

65 if props.correlation_data is not None: 65 ↛ 67line 65 didn't jump to line 67 because the condition on line 65 was always true(empty)

66 correlation_id = props.correlation_data.decode(errors="replace") (empty)

67 headers.update(props.user_properties) (empty)

68 

69 return MQTTMessage( (empty)

70 raw_message=msg, 

71 body=msg.payload, 

72 headers=headers, 

73 content_type=content_type, 

74 reply_to=reply_to, 

75 correlation_id=correlation_id, 

76 )