Coverage for amqtt/errors.py: 100%

1from typing import Any 

2 

3 

4class AMQTTError(Exception): 

5 """aMQTT base exception.""" 

6 

7 

8class MQTTError(Exception): 

9 """Base class for all errors referring to MQTT specifications.""" 

10 

11 

12class CodecError(Exception): 

13 """Exceptions thrown by packet encode/decode functions.""" 

14 

15 

16class NoDataError(Exception): 

17 """Exceptions thrown by packet encode/decode functions.""" 

18 

19 

20class ZeroLengthReadError(NoDataError): 

21 def __init__(self) -> None: 

22 super().__init__("Decoding a string of length zero.") 

23 

24 

25class BrokerError(Exception): 

26 """Exceptions thrown by broker.""" 

27 

28 

29class PluginError(Exception): 

30 """Exceptions thrown when loading or initializing a plugin.""" 

31 

32 

33class PluginImportError(PluginError): 

34 """Exceptions thrown when loading plugin.""" 

35 

36 

37class PluginCoroError(PluginError): 

38 """Exceptions thrown when loading a plugin with a non-async call method.""" 

39 

40 

41class PluginInitError(PluginError): 

42 """Exceptions thrown when initializing plugin.""" 

43 

44 def __init__(self, plugin: Any) -> None: 

45 super().__init__(f"Plugin init failed: {plugin!r}") 

46 

47 

48class ClientError(Exception): 

49 """Exceptions thrown by client.""" 

50 

51 

52class ConnectError(ClientError): 

53 """Exceptions thrown by client connect.""" 

54 

55 return_code: int | None = None 

56 

57 

58class ProtocolHandlerError(Exception): 

59 """Exceptions thrown by protocol handle.""" 

60 

61 

62class PluginLoadError(Exception): 

63 """Exception thrown when loading a plugin."""