Coverage for amqtt/mqtt/connack.py: 93%

1from typing_extensions import Self 

2 

3from amqtt.adapters import ReaderAdapter 

4from amqtt.codecs_amqtt import bytes_to_int, read_or_raise 

5from amqtt.errors import AMQTTError 

6from amqtt.mqtt.packet import CONNACK, MQTTFixedHeader, MQTTPacket, MQTTPayload, MQTTVariableHeader 

7 

8CONNECTION_ACCEPTED = 0x00 

9UNACCEPTABLE_PROTOCOL_VERSION = 0x01 

10IDENTIFIER_REJECTED = 0x02 

11SERVER_UNAVAILABLE = 0x03 

12BAD_USERNAME_PASSWORD = 0x04 

13NOT_AUTHORIZED = 0x05 

14 

15 

16class ConnackVariableHeader(MQTTVariableHeader): 

17 __slots__ = ("return_code", "session_parent") 

18 

19 def __init__(self, session_parent: int | None = None, return_code: int | None = None) -> None: 

20 super().__init__() 

21 self.session_parent = session_parent 

22 self.return_code = return_code 

23 

24 @classmethod 

25 async def from_stream(cls, reader: ReaderAdapter, _: MQTTFixedHeader | None) -> Self: 

26 data = await read_or_raise(reader, 2) 

27 session_parent = data[0] & 0x01 

28 return_code = bytes_to_int(data[1]) 

29 return cls(session_parent, return_code) 

30 

31 def to_bytes(self) -> bytes | bytearray: 

32 out = bytearray(2) 

33 # Connect acknowledge flags 

34 out[0] = 1 if self.session_parent else 0 

35 # Return code 

36 out[1] = self.return_code or 0 

37 return out 

38 

39 def __repr__(self) -> str: 

40 """Return a string representation of the ConnackVariableHeader object.""" 

41 return f"{type(self).__name__}(session_parent={hex(self.session_parent or 0)}, return_code={hex(self.return_code or 0)})" 

42 

43 

44class ConnackPacket(MQTTPacket[ConnackVariableHeader, MQTTPayload[MQTTVariableHeader], MQTTFixedHeader]): 

45 VARIABLE_HEADER = ConnackVariableHeader 

46 PAYLOAD = MQTTPayload[MQTTVariableHeader] 

47 

48 def __init__( 

49 self, 

50 fixed: MQTTFixedHeader | None = None, 

51 variable_header: ConnackVariableHeader | None = None, 

52 payload: MQTTPayload[MQTTVariableHeader] | None = None, 

53 ) -> None: 

54 if fixed is None: 

55 header = MQTTFixedHeader(CONNACK, 0x00) 

56 elif fixed.packet_type != CONNACK: 

57 msg = f"Invalid fixed packet type {fixed.packet_type} for ConnackPacket init" 

58 raise AMQTTError(msg) from None 

59 else: 

60 header = fixed 

61 

62 super().__init__(header, variable_header, payload) 

63 

64 @classmethod 

65 def build(cls, session_parent: int | None = None, return_code: int | None = None) -> Self: 

66 v_header = ConnackVariableHeader(session_parent, return_code) 

67 return cls(variable_header=v_header) 

68 

69 @property 

70 def return_code(self) -> int | None: 

71 if self.variable_header is None: 

72 msg = "Variable header is not set" 

73 raise ValueError(msg) 

74 return self.variable_header.return_code 

75 

76 @return_code.setter 

77 def return_code(self, return_code: int | None) -> None: 

78 if self.variable_header is None: 78 ↛ 81line 78 didn't jump to line 81 because the condition on line 78 was always true

79 msg = "Variable header is not set" 

80 raise ValueError(msg) 

81 self.variable_header.return_code = return_code 

82 

83 @property 

84 def session_parent(self) -> int | None: 

85 if self.variable_header is None: 85 ↛ 88line 85 didn't jump to line 88 because the condition on line 85 was always true

86 msg = "Variable header is not set" 

87 raise ValueError(msg) 

88 return self.variable_header.session_parent 

89 

90 @session_parent.setter 

91 def session_parent(self, session_parent: int | None) -> None: 

92 if self.variable_header is None: 92 ↛ 95line 92 didn't jump to line 95 because the condition on line 92 was always true

93 msg = "Variable header is not set" 

94 raise ValueError(msg) 

95 self.variable_header.session_parent = session_parent