Coverage for zenmqtt/mqtt/limits.py: 100%

1from dataclasses import dataclass 

2from typing import Final, Optional, Sequence 

3 

4from zenmqtt.exceptions import ( 

5 FeatureNotSupportedError, 

6 OutgoingPacketTooLargeError, 

7 QoSNotSupportedError, 

8) 

9from zenmqtt.mqtt.connect import ConnackProperties 

10from zenmqtt.mqtt.publish import PublishProperties 

11from zenmqtt.mqtt.subscribe import ( 

12 SubscriptionProperties, 

13 SubscriptionRequest, 

14 to_subscription, 

15) 

16 

17SHARED_SUBSCRIPTION_PREFIX: Final[str] = "$share/" 

18 

19# "Receive Maximum" of the client or the server if it isn't sent in CONNECT 

20# or CONNACK (MQTT 5, 3.1.2.11.3 and 3.2.2.3.3) 

21DEFAULT_RECEIVE_MAXIMUM: Final[int] = 2**16 - 1 

22 

23 

24@dataclass(frozen=True) 

25class ServerLimits: 

26 """ 

27 Limits of the server from CONNACK, absent properties have default values 

28 defined by the spec. 

29 """ 

30 

31 __slots__ = ( 

32 "maximum_qos", 

33 "retain_available", 

34 "maximum_packet_size", 

35 "topic_alias_maximum", 

36 "wildcard_subscription_available", 

37 "subscription_identifier_available", 

38 "shared_subscription_available", 

39 ) 

40 

41 maximum_qos: int 

42 retain_available: bool 

43 # None means no limit 

44 maximum_packet_size: Optional[int] 

45 # 0 means topic aliases aren't allowed 

46 topic_alias_maximum: int 

47 wildcard_subscription_available: bool 

48 subscription_identifier_available: bool 

49 shared_subscription_available: bool 

50 

51 @classmethod 

52 def from_connack(cls, properties: ConnackProperties) -> "ServerLimits": 

53 return cls( 

54 maximum_qos=properties.get("maximum_qos", 2), 

55 retain_available=properties.get("retain_available", True), 

56 maximum_packet_size=properties.get("maximum_packet_size"), 

57 topic_alias_maximum=properties.get("topic_alias_maximum", 0), 

58 wildcard_subscription_available=properties.get( 

59 "wildcard_subscription_available", True 

60 ), 

61 subscription_identifier_available=properties.get( 

62 "subscription_identifier_available", True 

63 ), 

64 shared_subscription_available=properties.get( 

65 "shared_subscription_available", True 

66 ), 

67 ) 

68 

69 def check_publish( 

70 self, qos: int, retain: bool, properties: PublishProperties 

71 ) -> None: 

72 if qos > self.maximum_qos: 

73 raise QoSNotSupportedError( 

74 f"QoS {qos} is higher than server's maximum QoS {self.maximum_qos}" 

75 ) 

76 

77 if retain and not self.retain_available: 

78 raise FeatureNotSupportedError( 

79 "The server doesn't support retained messages" 

80 ) 

81 

82 if (topic_alias := properties.get("topic_alias")) is not None: 

83 if not 1 <= topic_alias <= self.topic_alias_maximum: 

84 raise FeatureNotSupportedError( 

85 f"Topic alias {topic_alias} isn't in the range allowed by " 

86 f"the server: 1..{self.topic_alias_maximum}" 

87 ) 

88 

89 def check_subscribe( 

90 self, topics: Sequence[SubscriptionRequest], properties: SubscriptionProperties 

91 ) -> None: 

92 if ( 

93 "subscription_identifier" in properties 

94 and not self.subscription_identifier_available 

95 ): 

96 raise FeatureNotSupportedError( 

97 "The server doesn't support subscription identifiers" 

98 ) 

99 

100 for request in topics: 

101 topic = to_subscription(request).topic 

102 

103 if ( 

104 topic.startswith(SHARED_SUBSCRIPTION_PREFIX) 

105 and not self.shared_subscription_available 

106 ): 

107 raise FeatureNotSupportedError( 

108 f"The server doesn't support shared subscriptions: {topic}" 

109 ) 

110 

111 if ( 

112 "+" in topic or "#" in topic 

113 ) and not self.wildcard_subscription_available: 

114 raise FeatureNotSupportedError( 

115 f"The server doesn't support wildcard subscriptions: {topic}" 

116 ) 

117 

118 def check_packet_size(self, packet: bytes) -> None: 

119 if ( 

120 self.maximum_packet_size is not None 

121 and len(packet) > self.maximum_packet_size 

122 ): 

123 raise OutgoingPacketTooLargeError( 

124 f"Packet size {len(packet)} is bigger than server's maximum " 

125 f"packet size {self.maximum_packet_size}" 

126 ) 

127 

128 

129DEFAULT_SERVER_LIMITS: Final[ServerLimits] = ServerLimits.from_connack({})