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

1from dataclasses import dataclass 

2from typing import Sequence, Tuple, TypedDict, cast 

3 

4from zenmqtt.mqtt.packet import BytesReader, FixedHeader 

5from zenmqtt.mqtt.properties import parse_properties 

6from zenmqtt.mqtt.reason_codes import ReasonCode, to_reason_code 

7 

8 

9class AuthProperties(TypedDict, total=False): 

10 authentication_method: str 

11 authentication_data: bytes 

12 reason_string: str 

13 user_property: Sequence[Tuple[str, str]] 

14 

15 

16@dataclass(frozen=True) 

17class AuthResult: 

18 __slots__ = ("reason_code", "properties") 

19 

20 reason_code: ReasonCode 

21 properties: AuthProperties 

22 

23 

24def parse_auth_packet(fixed_header: FixedHeader, reader: BytesReader) -> AuthResult: 

25 # both the reason code and properties may be omitted 

26 reason_code = to_reason_code(reader.read_byte() if reader.remaining() else 0x0) 

27 properties = parse_properties(reader) if reader.remaining() else {} 

28 

29 reader.ensure_end() 

30 

31 return AuthResult( 

32 reason_code=reason_code, 

33 properties=cast(AuthProperties, properties), 

34 )