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

1from dataclasses import dataclass 

2 

3from zenmqtt.exceptions import MalformedPacketError 

4from zenmqtt.mqtt.packet import BytesReader, FixedHeader, PacketType 

5from zenmqtt.mqtt.utils import pack_fixed_header 

6 

7# PINGREQ and PINGRESP consist of the fixed header only: 

8# reserved flags are 0x0, no variable header and no payload 

9 

10 

11def pack_pingreq_packet() -> bytes: 

12 return pack_fixed_header(PacketType.PINGREQ, 0x00, 0) 

13 

14 

15@dataclass(frozen=True) 

16class PingRespResult: 

17 __slots__ = () 

18 

19 

20def parse_pingresp_packet( 

21 fixed_header: FixedHeader, reader: BytesReader 

22) -> PingRespResult: 

23 assert fixed_header.packet_type == PacketType.PINGRESP 

24 

25 if fixed_header.flags != 0x00 or fixed_header.length != 0: 

26 raise MalformedPacketError(f"Malformed PINGRESP packet: {fixed_header}") 

27 

28 return PingRespResult()