Coverage for amqtt/mqtt/__init__.py: 81%

1"""INIT.""" 

2 

3__all__ = ["MQTTPacket"] 

4 

5from typing import Any, TypeAlias 

6 

7from amqtt.errors import AMQTTError 

8from amqtt.mqtt.connack import ConnackPacket 

9from amqtt.mqtt.connect import ConnectPacket 

10from amqtt.mqtt.disconnect import DisconnectPacket 

11from amqtt.mqtt.packet import ( 

12 CONNACK, 

13 CONNECT, 

14 DISCONNECT, 

15 PINGREQ, 

16 PINGRESP, 

17 PUBACK, 

18 PUBCOMP, 

19 PUBLISH, 

20 PUBREC, 

21 PUBREL, 

22 SUBACK, 

23 SUBSCRIBE, 

24 UNSUBACK, 

25 UNSUBSCRIBE, 

26 MQTTFixedHeader, 

27 MQTTPacket, 

28) 

29from amqtt.mqtt.pingreq import PingReqPacket 

30from amqtt.mqtt.pingresp import PingRespPacket 

31from amqtt.mqtt.puback import PubackPacket 

32from amqtt.mqtt.pubcomp import PubcompPacket 

33from amqtt.mqtt.publish import PublishPacket 

34from amqtt.mqtt.pubrec import PubrecPacket 

35from amqtt.mqtt.pubrel import PubrelPacket 

36from amqtt.mqtt.suback import SubackPacket 

37from amqtt.mqtt.subscribe import SubscribePacket 

38from amqtt.mqtt.unsuback import UnsubackPacket 

39from amqtt.mqtt.unsubscribe import UnsubscribePacket 

40 

41_P: TypeAlias = MQTTPacket[Any, Any, Any] 

42 

43packet_dict: dict[int, type[_P]] = { 

44 CONNECT: ConnectPacket, 

45 CONNACK: ConnackPacket, 

46 PUBLISH: PublishPacket, 

47 PUBACK: PubackPacket, 

48 PUBREC: PubrecPacket, 

49 PUBREL: PubrelPacket, 

50 PUBCOMP: PubcompPacket, 

51 SUBSCRIBE: SubscribePacket, 

52 SUBACK: SubackPacket, 

53 UNSUBSCRIBE: UnsubscribePacket, 

54 UNSUBACK: UnsubackPacket, 

55 PINGREQ: PingReqPacket, 

56 PINGRESP: PingRespPacket, 

57 DISCONNECT: DisconnectPacket, 

58} 

59 

60 

61def packet_class(fixed_header: MQTTFixedHeader) -> type[_P]: 

62 """Return the packet class for a given fixed header. 

63 

64 :param fixed_header: The fixed header of the packet. 

65 :type 

66 fixed_header: MQTTFixedHeader 

67 :return: The packet class for the given fixed header. 

68 :rtype: type[MQTTPacket] 

69 :raises AMQTTError: If the packet type is not recognized. 

70 """ 

71 if fixed_header.packet_type not in packet_dict: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true

72 msg = f"Unexpected packet Type '{fixed_header.packet_type}'" 

73 raise AMQTTError(msg) 

74 try: 

75 return packet_dict[fixed_header.packet_type] 

76 except KeyError as e: 

77 msg = f"Unexpected packet Type '{fixed_header.packet_type}'" 

78 raise AMQTTError(msg) from e