1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
|
import paho.mqtt.client as mqtt
import pytest
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.reasoncodes import ReasonCode
from aiomqtt.exceptions import _CONNECT_RC_STRINGS, MqttCodeError, MqttConnectError
@pytest.mark.parametrize(
"rc",
[
mqtt.MQTT_ERR_SUCCESS,
mqtt.MQTT_ERR_NOMEM,
mqtt.MQTT_ERR_PROTOCOL,
mqtt.MQTT_ERR_INVAL,
mqtt.MQTT_ERR_NO_CONN,
mqtt.MQTT_ERR_CONN_REFUSED,
mqtt.MQTT_ERR_NOT_FOUND,
mqtt.MQTT_ERR_CONN_LOST,
mqtt.MQTT_ERR_TLS,
mqtt.MQTT_ERR_PAYLOAD_SIZE,
mqtt.MQTT_ERR_NOT_SUPPORTED,
mqtt.MQTT_ERR_AUTH,
mqtt.MQTT_ERR_ACL_DENIED,
mqtt.MQTT_ERR_UNKNOWN,
mqtt.MQTT_ERR_ERRNO,
mqtt.MQTT_ERR_QUEUE_SIZE,
mqtt.MQTT_ERR_KEEPALIVE,
-1,
],
)
def test_mqtt_code_error_int(rc: int) -> None:
assert str(MqttCodeError(rc)) == f"[code:{rc}] {mqtt.error_string(rc)}"
@pytest.mark.parametrize(
"packet_type, a_name",
[
(PacketTypes.CONNACK, "Success"),
(PacketTypes.PUBACK, "Success"),
(PacketTypes.SUBACK, "Granted QoS 1"),
],
)
def test_mqtt_code_error_reason_codes(packet_type: int, a_name: str) -> None:
rc = ReasonCode(packet_type, a_name)
assert str(MqttCodeError(rc)) == f"[code:{rc.value}] {rc!s}"
def test_mqtt_code_error_none() -> None:
assert str(MqttCodeError(None)) == "[code:None] "
@pytest.mark.parametrize("rc, message", [*_CONNECT_RC_STRINGS.items(), (0, "")])
def test_mqtt_connect_error_int(rc: int, message: str) -> None:
error = MqttConnectError(rc)
arg = "Connection refused"
if rc in _CONNECT_RC_STRINGS:
arg += f": {message}"
assert error.args[0] == arg
assert str(error) == f"[code:{rc}] {mqtt.error_string(rc)}"
@pytest.mark.parametrize(
"packet_type, a_name",
[
(PacketTypes.CONNACK, "Success"),
(PacketTypes.PUBACK, "Success"),
(PacketTypes.SUBACK, "Granted QoS 1"),
],
)
def test_mqtt_connect_error_reason_codes(packet_type: int, a_name: str) -> None:
rc = ReasonCode(packet_type, a_name)
assert str(MqttConnectError(rc)) == f"[code:{rc.value}] {rc!s}"
|
