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

1import struct 

2 

3from zenmqtt.mqtt.packet import PacketType 

4 

5 

6def pack_variable_byte_integer(value: int) -> bytearray: 

7 result = bytearray() 

8 

9 while True: 

10 value, remainder = divmod(value, 0x80) 

11 

12 if value > 0: 

13 remainder |= 0x80 

14 

15 result.extend(struct.pack("!B", remainder)) 

16 

17 if value <= 0: 

18 break 

19 

20 return result 

21 

22 

23def pack_str16(value: str) -> bytes: 

24 encoded = value.encode() 

25 return pack_binary(encoded) 

26 

27 

28def pack_binary(value: bytes) -> bytes: 

29 return struct.pack(f"!H{len(value)}s", len(value), value) 

30 

31 

32def pack_fixed_header(packet_type: PacketType, flags: int, length: int) -> bytes: 

33 result = bytearray() 

34 

35 result.extend(struct.pack("!B", (packet_type << 4) | flags)) 

36 result.extend(pack_variable_byte_integer(length)) 

37 

38 return bytes(result)