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

1from enum import IntEnum 

2from typing import Final, Optional 

3 

4 

5class ReasonCode(IntEnum): 

6 """ 

7 Reason codes of MQTT 5.0 (section 2.4). A value is an int, so it can be 

8 passed where a reason code is expected and compared with ints. 

9 

10 Some values have several names in the specification, depending on the 

11 packet: 0x00 is "Success", "Normal disconnection" and "Granted QoS 0". 

12 They are aliases of one member, which is shown as SUCCESS. 

13 

14 A value which isn't in the specification, e.g. sent by a broker which 

15 breaks it, gives a member named like UNKNOWN_0x42 instead of ValueError. 

16 """ 

17 

18 SUCCESS = 0x00 

19 NORMAL_DISCONNECTION = 0x00 

20 GRANTED_QOS_0 = 0x00 

21 GRANTED_QOS_1 = 0x01 

22 GRANTED_QOS_2 = 0x02 

23 DISCONNECT_WITH_WILL_MESSAGE = 0x04 

24 NO_MATCHING_SUBSCRIBERS = 0x10 

25 NO_SUBSCRIPTION_EXISTED = 0x11 

26 CONTINUE_AUTHENTICATION = 0x18 

27 RE_AUTHENTICATE = 0x19 

28 UNSPECIFIED_ERROR = 0x80 

29 MALFORMED_PACKET = 0x81 

30 PROTOCOL_ERROR = 0x82 

31 IMPLEMENTATION_SPECIFIC_ERROR = 0x83 

32 UNSUPPORTED_PROTOCOL_VERSION = 0x84 

33 CLIENT_IDENTIFIER_NOT_VALID = 0x85 

34 BAD_USER_NAME_OR_PASSWORD = 0x86 

35 NOT_AUTHORIZED = 0x87 

36 SERVER_UNAVAILABLE = 0x88 

37 SERVER_BUSY = 0x89 

38 BANNED = 0x8A 

39 SERVER_SHUTTING_DOWN = 0x8B 

40 BAD_AUTHENTICATION_METHOD = 0x8C 

41 KEEP_ALIVE_TIMEOUT = 0x8D 

42 SESSION_TAKEN_OVER = 0x8E 

43 TOPIC_FILTER_INVALID = 0x8F 

44 TOPIC_NAME_INVALID = 0x90 

45 PACKET_IDENTIFIER_IN_USE = 0x91 

46 PACKET_IDENTIFIER_NOT_FOUND = 0x92 

47 RECEIVE_MAXIMUM_EXCEEDED = 0x93 

48 TOPIC_ALIAS_INVALID = 0x94 

49 PACKET_TOO_LARGE = 0x95 

50 MESSAGE_RATE_TOO_HIGH = 0x96 

51 QUOTA_EXCEEDED = 0x97 

52 ADMINISTRATIVE_ACTION = 0x98 

53 PAYLOAD_FORMAT_INVALID = 0x99 

54 RETAIN_NOT_SUPPORTED = 0x9A 

55 QOS_NOT_SUPPORTED = 0x9B 

56 USE_ANOTHER_SERVER = 0x9C 

57 SERVER_MOVED = 0x9D 

58 SHARED_SUBSCRIPTIONS_NOT_SUPPORTED = 0x9E 

59 CONNECTION_RATE_EXCEEDED = 0x9F 

60 MAXIMUM_CONNECT_TIME = 0xA0 

61 SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED = 0xA1 

62 WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED = 0xA2 

63 

64 @property 

65 def is_failure(self) -> bool: 

66 """Reason codes from 0x80 indicate failure.""" 

67 return self >= FAILURE_REASON_CODE 

68 

69 @classmethod 

70 def _missing_(cls, value: object) -> Optional["ReasonCode"]: 

71 if not isinstance(value, int) or not 0 <= value <= 0xFF: 

72 return None 

73 

74 member = int.__new__(cls, value) 

75 member._name_ = f"UNKNOWN_0x{value:02X}" 

76 member._value_ = value 

77 return member 

78 

79 

80# every byte value: a reason code read from a packet is turned into the enum by 

81# indexing, several times faster than ReasonCode(value); unknown values get one 

82# member each 

83_BY_BYTE: Final[tuple[ReasonCode, ...]] = tuple( 

84 ReasonCode(value) for value in range(0x100) 

85) 

86 

87 

88def to_reason_code(value: int) -> ReasonCode: 

89 """A reason code read from a packet (one byte).""" 

90 return _BY_BYTE[value] 

91 

92 

93def to_reason_codes(data: bytes) -> list[ReasonCode]: 

94 """Reason codes read from a packet, one per byte.""" 

95 return list(map(_BY_BYTE.__getitem__, data)) 

96 

97 

98# reason codes >= 0x80 indicate failure 

99FAILURE_REASON_CODE: Final[int] = 0x80 

100# DISCONNECT reason code when a valid packet can't be processed, e.g. the 

101# session storage failed 

102IMPLEMENTATION_SPECIFIC_ERROR: Final[int] = 0x83 

103# PUBREL/PUBCOMP reason code for an unknown packet identifier 

104PACKET_IDENTIFIER_NOT_FOUND: Final[int] = 0x92