add unit test for recv timeout

Author: vladak

tests/test_recv_timeout.py

@@ -0,0 +1,53 @@
+# 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"""
+        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=lambda ret_buf, buf_size: 0)
+        # 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()