Coverage for amqtt/mqtt/unsuback.py: 88%

1from typing_extensions import Self 

2 

3from amqtt.errors import AMQTTError 

4from amqtt.mqtt.packet import UNSUBACK, MQTTFixedHeader, MQTTPacket, PacketIdVariableHeader 

5 

6 

7class UnsubackPacket(MQTTPacket[PacketIdVariableHeader, None, MQTTFixedHeader]): 

8 VARIABLE_HEADER = PacketIdVariableHeader 

9 PAYLOAD = None 

10 

11 def __init__( 

12 self, 

13 fixed: MQTTFixedHeader | None = None, 

14 variable_header: PacketIdVariableHeader | None = None, 

15 payload: None = None, 

16 ) -> None: 

17 if fixed is None: 

18 header = MQTTFixedHeader(UNSUBACK, 0x00) 

19 else: 

20 if fixed.packet_type is not UNSUBACK: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 msg = f"Invalid fixed packet type {fixed.packet_type} for UnsubackPacket init" 

22 raise AMQTTError( 

23 msg, 

24 ) 

25 header = fixed 

26 

27 super().__init__(header) 

28 self.variable_header = variable_header 

29 self.payload = payload 

30 

31 @classmethod 

32 def build(cls, packet_id: int) -> Self: 

33 variable_header = PacketIdVariableHeader(packet_id) 

34 return cls(variable_header=variable_header)