Merge branch 'main' into timestamp_ns

Author: kevin-tritz

adafruit_minimqtt/adafruit_minimqtt.py

@@ -73,12 +73,18 @@
 MQTT_PKT_TYPE_MASK = const(0xF0)
 
 
+CONNACK_ERROR_INCORRECT_PROTOCOL = const(0x01)
+CONNACK_ERROR_ID_REJECTED = const(0x02)
+CONNACK_ERROR_SERVER_UNAVAILABLE = const(0x03)
+CONNACK_ERROR_INCORECT_USERNAME_PASSWORD = const(0x04)
+CONNACK_ERROR_UNAUTHORIZED = const(0x05)
+
 CONNACK_ERRORS = {
-    const(0x01): "Connection Refused - Incorrect Protocol Version",
-    const(0x02): "Connection Refused - ID Rejected",
-    const(0x03): "Connection Refused - Server unavailable",
-    const(0x04): "Connection Refused - Incorrect username/password",
-    const(0x05): "Connection Refused - Unauthorized",
+    CONNACK_ERROR_INCORRECT_PROTOCOL: "Connection Refused - Incorrect Protocol Version",
+    CONNACK_ERROR_ID_REJECTED: "Connection Refused - ID Rejected",
+    CONNACK_ERROR_SERVER_UNAVAILABLE: "Connection Refused - Server unavailable",
+    CONNACK_ERROR_INCORECT_USERNAME_PASSWORD: "Connection Refused - Incorrect username/password",
+    CONNACK_ERROR_UNAUTHORIZED: "Connection Refused - Unauthorized",
 }
 
 _default_sock = None  # pylint: disable=invalid-name
@@ -88,6 +94,10 @@
 class MMQTTException(Exception):
     """MiniMQTT Exception class."""
 
+    def __init__(self, error, code=None):
+        super().__init__(error, code)
+        self.code = code
+
 
 class NullLogger:
     """Fake logger class that does not do anything"""
@@ -397,21 +407,31 @@ def connect(
                 )
                 self._reset_reconnect_backoff()
                 return ret
-            except RuntimeError as e:
+            except (MemoryError, OSError, RuntimeError) as e:
+                if isinstance(e, RuntimeError) and e.args == ("pystack exhausted",):
+                    raise
                 self.logger.warning(f"Socket error when connecting: {e}")
+                last_exception = e
                 backoff = False
             except MMQTTException as e:
-                last_exception = e
+                self._close_socket()
                 self.logger.info(f"MMQT error: {e}")
+                if e.code in [
+                    CONNACK_ERROR_INCORECT_USERNAME_PASSWORD,
+                    CONNACK_ERROR_UNAUTHORIZED,
+                ]:
+                    # No sense trying these again, re-raise
+                    raise
+                last_exception = e
                 backoff = True
 
         if self._reconnect_attempts_max > 1:
             exc_msg = "Repeated connect failures"
         else:
             exc_msg = "Connect failure"
+
         if last_exception:
             raise MMQTTException(exc_msg) from last_exception
-
         raise MMQTTException(exc_msg)
 
     # pylint: disable=too-many-branches, too-many-statements, too-many-locals
@@ -508,7 +528,7 @@ def _connect(
                 rc = self._sock_exact_recv(3)
                 assert rc[0] == 0x02
                 if rc[2] != 0x00:
-                    raise MMQTTException(CONNACK_ERRORS[rc[2]])
+                    raise MMQTTException(CONNACK_ERRORS[rc[2]], code=rc[2])
                 self._is_connected = True
                 result = rc[0] & 1
                 if self.on_connect is not None:
@@ -522,6 +542,12 @@ def _connect(
                         f"No data received from broker for {self._recv_timeout} seconds."
                     )
 
+    def _close_socket(self):
+        if self._sock:
+            self.logger.debug("Closing socket")
+            self._connection_manager.close_socket(self._sock)
+            self._sock = None
+
     # pylint: disable=no-self-use
     def _encode_remaining_length(
         self, fixed_header: bytearray, remaining_length: int
@@ -550,8 +576,7 @@ def disconnect(self) -> None:
             self._sock.send(MQTT_DISCONNECT)
         except RuntimeError as e:
             self.logger.warning(f"Unable to send DISCONNECT packet: {e}")
-        self.logger.debug("Closing socket")
-        self._connection_manager.close_socket(self._sock)
+        self._close_socket()
         self._is_connected = False
         self._subscribed_topics = []
         self._last_msg_sent_timestamp = 0
@@ -568,6 +593,7 @@ def ping(self) -> list[int]:
         self._sock.send(MQTT_PINGREQ)
         ping_timeout = self.keep_alive
         stamp = ticks_ms()
+
         self._last_msg_sent_timestamp = stamp
         rc, rcs = None, []
         while rc != MQTT_PINGRESP:
@@ -946,7 +972,7 @@ def _wait_for_msg(self, timeout: Optional[float] = None) -> Optional[int]:
                 res = self._sock_exact_recv(1)
             except self._socket_pool.timeout:
                 return None
-        else:  # socketpool, esp32spi
+        else:  # socketpool, esp32spi, wiznet5k
             try:
                 res = self._sock_exact_recv(1, timeout=timeout)
             except OSError as error:
@@ -1035,14 +1061,14 @@ def _sock_exact_recv(
         """
         stamp = ticks_ms()
         if not self._backwards_compatible_sock:
-            # CPython/Socketpool Impl.
+            # CPython, socketpool, esp32spi, wiznet5k
             rc = bytearray(bufsize)
             mv = memoryview(rc)
             recv_len = self._sock.recv_into(rc, bufsize)
             to_read = bufsize - recv_len
             if to_read < 0:
                 raise MMQTTException(f"negative number of bytes to read: {to_read}")
-            read_timeout = timeout if timeout is not None else self.keep_alive
+            read_timeout = timeout if timeout is not None else self._recv_timeout
             mv = mv[recv_len:]
             while to_read > 0:
                 recv_len = self._sock.recv_into(mv, to_read)
@@ -1052,8 +1078,8 @@ def _sock_exact_recv(
                     raise MMQTTException(
                         f"Unable to receive {to_read} bytes within {read_timeout} seconds."
                     )
-        else:  # ESP32SPI Impl.
-            # This will timeout with socket timeout (not keepalive timeout)
+        else:  # Legacy: fona, esp_atcontrol
+            # This will time out with socket timeout (not receive timeout).
             rc = self._sock.recv(bufsize)
             if not rc:
                 self.logger.debug("_sock_exact_recv timeout")
@@ -1063,7 +1089,7 @@ def _sock_exact_recv(
             # or raise exception if wait longer than read_timeout
             to_read = bufsize - len(rc)
             assert to_read >= 0
-            read_timeout = self.keep_alive
+            read_timeout = self._recv_timeout
             while to_read > 0:
                 recv = self._sock.recv(to_read)
                 to_read -= len(recv)

tests/test_backoff.py

@@ -18,18 +18,24 @@ class TestExpBackOff:
     """basic exponential back-off test"""
 
     connect_times = []
+    raise_exception = None
 
     # pylint: disable=unused-argument
     def fake_connect(self, arg):
         """connect() replacement that records the call times and always raises OSError"""
         self.connect_times.append(time.monotonic())
-        raise OSError("this connect failed")
+        raise self.raise_exception
 
     def test_failing_connect(self) -> None:
         """test that exponential back-off is used when connect() always raises OSError"""
         # use RFC 1918 address to avoid dealing with IPv6 in the call list below
         host = "172.40.0.3"
         port = 1883
+        self.connect_times = []
+        error_code = MQTT.CONNACK_ERROR_SERVER_UNAVAILABLE
+        self.raise_exception = MQTT.MMQTTException(
+            MQTT.CONNACK_ERRORS[error_code], code=error_code
+        )
 
         with patch.object(socket.socket, "connect") as mock_method:
             mock_method.side_effect = self.fake_connect
@@ -45,6 +51,7 @@ def test_failing_connect(self) -> None:
             print("connecting")
             with pytest.raises(MQTT.MMQTTException) as context:
                 mqtt_client.connect()
+                assert mqtt_client._sock is None
             assert "Repeated connect failures" in str(context)
 
             mock_method.assert_called()
@@ -54,3 +61,34 @@ def test_failing_connect(self) -> None:
             print(f"connect() call times: {self.connect_times}")
             for i in range(1, connect_retries):
                 assert self.connect_times[i] >= 2**i
+
+    def test_unauthorized(self) -> None:
+        """test that exponential back-off is used when connect() always raises OSError"""
+        # use RFC 1918 address to avoid dealing with IPv6 in the call list below
+        host = "172.40.0.3"
+        port = 1883
+        self.connect_times = []
+        error_code = MQTT.CONNACK_ERROR_UNAUTHORIZED
+        self.raise_exception = MQTT.MMQTTException(
+            MQTT.CONNACK_ERRORS[error_code], code=error_code
+        )
+
+        with patch.object(socket.socket, "connect") as mock_method:
+            mock_method.side_effect = self.fake_connect
+
+            connect_retries = 3
+            mqtt_client = MQTT.MQTT(
+                broker=host,
+                port=port,
+                socket_pool=socket,
+                ssl_context=ssl.create_default_context(),
+                connect_retries=connect_retries,
+            )
+            print("connecting")
+            with pytest.raises(MQTT.MMQTTException) as context:
+                mqtt_client.connect()
+                assert mqtt_client._sock is None
+            assert "Connection Refused - Unauthorized" in str(context)
+
+            mock_method.assert_called()
+            assert len(self.connect_times) == 1

tests/test_port_ssl.py

@@ -120,6 +120,6 @@ def test_tls_without_ssl_context(self) -> None:
             connect_retries=1,
         )
 
-        with pytest.raises(AttributeError) as context:
+        with pytest.raises(ValueError) as context:
             mqtt_client.connect()
-        assert "ssl_context must be set" in str(context)
+        assert "ssl_context must be provided if using ssl" in str(context)

tests/test_recv_timeout.py

@@ -0,0 +1,56 @@
+# SPDX-FileCopyrightText: 2024 Vladimír Kotal
+#
+# SPDX-License-Identifier: Unlicense
+
+"""receive timeout tests"""
+
+import socket
+import time
+from unittest import TestCase, main
+from unittest.mock import Mock
+
+import adafruit_minimqtt.adafruit_minimqtt as MQTT
+
+
+class RecvTimeout(TestCase):
+    """This class contains tests for receive timeout handling."""
+
+    def test_recv_timeout_vs_keepalive(self) -> None:
+        """verify that receive timeout as used via ping() is different to keep alive timeout"""
+
+        for side_effect in [lambda ret_buf, buf_size: 0, socket.timeout]:
+            with self.subTest():
+                host = "127.0.0.1"
+
+                recv_timeout = 4
+                keep_alive = recv_timeout * 2
+                mqtt_client = MQTT.MQTT(
+                    broker=host,
+                    socket_pool=socket,
+                    connect_retries=1,
+                    socket_timeout=recv_timeout // 2,
+                    recv_timeout=recv_timeout,
+                    keep_alive=keep_alive,
+                )
+
+                # Create a mock socket that will accept anything and return nothing.
+                socket_mock = Mock()
+                socket_mock.recv_into = Mock(side_effect=side_effect)
+                # pylint: disable=protected-access
+                mqtt_client._sock = socket_mock
+
+                mqtt_client._connected = lambda: True
+                start = time.monotonic()
+                with self.assertRaises(MQTT.MMQTTException):
+                    mqtt_client.ping()
+
+                # Verify the mock interactions.
+                socket_mock.send.assert_called_once()
+                socket_mock.recv_into.assert_called()
+
+                now = time.monotonic()
+                assert recv_timeout <= (now - start) < keep_alive
+
+
+if __name__ == "__main__":
+    main()