Retry timeouts

test/__init__.py

@@ -10,9 +10,8 @@
 
 from awscrt import NativeResource
 from awscrt._test import check_for_leaks
-from awscrt.io import init_logging, LogLevel
+import time
 import unittest
-import sys
 
 TIMEOUT = 30.0
 
@@ -57,3 +56,23 @@ def tearDown(self):
         except Exception:
             NativeResourceTest._previous_test_failed = True
             raise
+
+
+MAX_RETRIES = 5
+
+
+def _is_retryable_exception(e):
+    exception_text = str(e)
+    return "AWS_IO_TLS_NEGOTIATION_TIMEOUT" in exception_text or "AWS_IO_SOCKET_TIMEOUT" in exception_text
+
+
+def test_retry_wrapper(test_function):
+    for i in range(MAX_RETRIES):
+        try:
+            test_function()
+            return
+        except Exception as e:
+            if _is_retryable_exception(e) and i + 1 < MAX_RETRIES:
+                time.sleep(1)
+            else:
+                raise

test/test_mqtt.py

@@ -4,7 +4,7 @@
 from awscrt.io import ClientBootstrap, ClientTlsContext, DefaultHostResolver, EventLoopGroup, Pkcs11Lib, TlsContextOptions
 from awscrt import http
 from awscrt.mqtt import Client, Connection, QoS, Will, OnConnectionClosedData, OnConnectionFailureData, OnConnectionSuccessData, ConnectReturnCode
-from test import NativeResourceTest
+from test import test_retry_wrapper, NativeResourceTest
 from concurrent.futures import Future
 import os
 import unittest
@@ -59,7 +59,7 @@ def _create_connection(
             on_connection_resumed=on_connection_resumed_callback)
         return connection
 
-    def test_connect_disconnect(self):
+    def _test_connect_disconnect(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
         test_input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_KEY")
@@ -71,7 +71,10 @@ def test_connect_disconnect(self):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_ecc_connect_disconnect(self):
+    def test_connect_disconnect(self):
+        test_retry_wrapper(self._test_connect_disconnect)
+
+    def _test_ecc_connect_disconnect(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_ECC_CERT")
         test_input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_ECC_KEY")
@@ -83,7 +86,10 @@ def test_ecc_connect_disconnect(self):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_pkcs11(self):
+    def test_ecc_connect_disconnect(self):
+        test_retry_wrapper(self._test_ecc_connect_disconnect)
+
+    def _test_pkcs11(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_pkcs11_lib = _get_env_variable("AWS_TEST_PKCS11_LIB")
         test_input_pkcs11_pin = _get_env_variable("AWS_TEST_PKCS11_PIN")
@@ -105,7 +111,10 @@ def test_pkcs11(self):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_pub_sub(self):
+    def test_pkcs11(self):
+        test_retry_wrapper(self._test_pkcs11)
+
+    def _test_pub_sub(self):
         self.TEST_TOPIC = '/test/me/senpai/' + str(uuid.uuid4())
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
@@ -148,7 +157,10 @@ def on_message(**kwargs):
         # disconnect
         connection.disconnect().result(TIMEOUT)
 
-    def test_will(self):
+    def test_pub_sub(self):
+        test_retry_wrapper(self._test_pub_sub)
+
+    def _test_will(self):
         self.TEST_TOPIC = '/test/me/senpai/' + str(uuid.uuid4())
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
@@ -239,7 +251,10 @@ def on_message(**kwargs):
         # disconnect
         subscriber.disconnect().result(TIMEOUT)
 
-    def test_on_message(self):
+    def test_will(self):
+        test_retry_wrapper(self._test_will)
+
+    def _test_on_message(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
         test_input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_KEY")
@@ -275,7 +290,10 @@ def on_message(**kwargs):
         # disconnect
         connection.disconnect().result(TIMEOUT)
 
-    def test_on_message_old_fn_signature(self):
+    def test_on_message(self):
+        test_retry_wrapper(self._test_on_message)
+
+    def _test_on_message_old_fn_signature(self):
         # ensure that message-received callbacks with the old function signature still work
 
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
@@ -320,7 +338,10 @@ def on_sub_message(topic, payload):
         # disconnect
         connection.disconnect().result(TIMEOUT)
 
-    def test_connect_disconnect_with_default_singletons(self):
+    def test_on_message_old_fn_signature(self):
+        test_retry_wrapper(self._test_on_message_old_fn_signature)
+
+    def _test_connect_disconnect_with_default_singletons(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
         test_input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_KEY")
@@ -337,7 +358,10 @@ def test_connect_disconnect_with_default_singletons(self):
         EventLoopGroup.release_static_default()
         DefaultHostResolver.release_static_default()
 
-    def test_connect_publish_wait_statistics_disconnect(self):
+    def test_connect_disconnect_with_default_singletons(self):
+        test_retry_wrapper(self._test_connect_disconnect_with_default_singletons)
+
+    def _test_connect_publish_wait_statistics_disconnect(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
         test_input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_KEY")
@@ -369,7 +393,10 @@ def test_connect_publish_wait_statistics_disconnect(self):
         # disconnect
         connection.disconnect().result(TIMEOUT)
 
-    def test_connect_publish_statistics_wait_disconnect(self):
+    def test_connect_publish_wait_statistics_disconnect(self):
+        test_retry_wrapper(self._test_connect_publish_wait_statistics_disconnect)
+
+    def _test_connect_publish_statistics_wait_disconnect(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
         test_input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_KEY")
@@ -409,7 +436,10 @@ def test_connect_publish_statistics_wait_disconnect(self):
         # disconnect
         connection.disconnect().result(TIMEOUT)
 
-    def test_connect_disconnect_with_callbacks_happy(self):
+    def test_connect_publish_statistics_wait_disconnect(self):
+        test_retry_wrapper(self._test_connect_publish_statistics_wait_disconnect)
+
+    def _test_connect_disconnect_with_callbacks_happy(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
         test_input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_KEY")
@@ -442,7 +472,10 @@ def on_connection_closed_callback(connection, callback_data: OnConnectionClosedD
         connection.disconnect().result(TIMEOUT)
         on_connection_closed_future.result(TIMEOUT)
 
-    def test_connect_disconnect_with_callbacks_unhappy(self):
+    def test_connect_disconnect_with_callbacks_happy(self):
+        test_retry_wrapper(self._test_connect_disconnect_with_callbacks_happy)
+
+    def _test_connect_disconnect_with_callbacks_unhappy(self):
         test_input_endpoint = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
         test_input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
         test_input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_KEY")
@@ -478,7 +511,10 @@ def on_connection_closed_callback(connection, callback_data: OnConnectionClosedD
         failure_data = on_onnection_failure_future.result(TIMEOUT)
         self.assertTrue(failure_data['error'] is not None)
 
-    def test_connect_disconnect_with_callbacks_happy_on_resume(self):
+    def test_connect_disconnect_with_callbacks_unhappy(self):
+        test_retry_wrapper(self._test_connect_disconnect_with_callbacks_unhappy)
+
+    def _test_connect_disconnect_with_callbacks_happy_on_resume(self):
         # Check that an on_connection_success callback fires on a resumed connection.
 
         # NOTE Since there is no mocked server available on this abstraction level, the only sensible approach
@@ -517,6 +553,10 @@ def on_connection_resumed_callback(connection, return_code: ConnectReturnCode, s
         self.assertEqual(success_data['return_code'], ConnectReturnCode.ACCEPTED)
         self.assertEqual(success_data['session_present'], False)
 
+        # Putting a sleep here helps prevent a "race" condition in IoT Core where the second connection can get
+        # rejected rather than the first disconnected.
+        time.sleep(5)
+
         # Reset the future for the reconnect attempt.
         on_connection_success_future = Future()
 
@@ -547,11 +587,14 @@ def on_connection_success_callback_dup(connection, callback_data: OnConnectionSu
         connection.disconnect().result(TIMEOUT)
         on_connection_closed_future.result(TIMEOUT)
 
+    def test_connect_disconnect_with_callbacks_happy_on_resume(self):
+        test_retry_wrapper(self._test_connect_disconnect_with_callbacks_happy_on_resume)
+
     # ==============================================================
     #             MOSQUITTO CONNECTION TESTS
     # ==============================================================
 
-    def test_mqtt311_direct_connect_minimum(self):
+    def _test_mqtt311_direct_connect_minimum(self):
         input_host_name = _get_env_variable("AWS_TEST_MQTT311_DIRECT_MQTT_HOST")
         input_port = int(_get_env_variable("AWS_TEST_MQTT311_DIRECT_MQTT_PORT"))
 
@@ -567,7 +610,10 @@ def test_mqtt311_direct_connect_minimum(self):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_mqtt311_direct_connect_basic_auth(self):
+    def test_mqtt311_direct_connect_minimum(self):
+        test_retry_wrapper(self._test_mqtt311_direct_connect_minimum)
+
+    def _test_mqtt311_direct_connect_basic_auth(self):
         input_host_name = _get_env_variable("AWS_TEST_MQTT311_DIRECT_MQTT_BASIC_AUTH_HOST")
         input_port = int(_get_env_variable("AWS_TEST_MQTT311_DIRECT_MQTT_BASIC_AUTH_PORT"))
         input_username = _get_env_variable("AWS_TEST_MQTT311_BASIC_AUTH_USERNAME")
@@ -587,7 +633,10 @@ def test_mqtt311_direct_connect_basic_auth(self):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_mqtt311_direct_connect_tls(self):
+    def test_mqtt311_direct_connect_basic_auth(self):
+        test_retry_wrapper(self._test_mqtt311_direct_connect_basic_auth)
+
+    def _test_mqtt311_direct_connect_tls(self):
         input_host_name = _get_env_variable("AWS_TEST_MQTT311_DIRECT_MQTT_TLS_HOST")
         input_port = int(_get_env_variable("AWS_TEST_MQTT311_DIRECT_MQTT_TLS_PORT"))
 
@@ -605,7 +654,10 @@ def test_mqtt311_direct_connect_tls(self):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_mqtt311_direct_connect_mutual_tls(self):
+    def test_mqtt311_direct_connect_tls(self):
+        test_retry_wrapper(self._test_mqtt311_direct_connect_tls)
+
+    def _test_mqtt311_direct_connect_mutual_tls(self):
         input_cert = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_CERT")
         input_key = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_RSA_KEY")
         input_host = _get_env_variable("AWS_TEST_MQTT311_IOT_CORE_HOST")
@@ -626,7 +678,10 @@ def test_mqtt311_direct_connect_mutual_tls(self):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_mqtt311_direct_connect_http_proxy_tls(self):
+    def test_mqtt311_direct_connect_mutual_tls(self):
+        test_retry_wrapper(self._test_mqtt311_direct_connect_mutual_tls)
+
+    def _test_mqtt311_direct_connect_http_proxy_tls(self):
         input_proxy_host = _get_env_variable("AWS_TEST_MQTT311_PROXY_HOST")
         input_proxy_port = int(_get_env_variable("AWS_TEST_MQTT311_PROXY_PORT"))
         input_host_name = _get_env_variable("AWS_TEST_MQTT311_DIRECT_MQTT_TLS_HOST")
@@ -655,7 +710,10 @@ def test_mqtt311_direct_connect_http_proxy_tls(self):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_mqtt311_websocket_connect_minimum(self):
+    def test_mqtt311_direct_connect_http_proxy_tls(self):
+        test_retry_wrapper(self._test_mqtt311_direct_connect_http_proxy_tls)
+
+    def _test_mqtt311_websocket_connect_minimum(self):
         input_host_name = _get_env_variable("AWS_TEST_MQTT311_WS_MQTT_HOST")
         input_port = int(_get_env_variable("AWS_TEST_MQTT311_WS_MQTT_PORT"))
 
@@ -677,7 +735,10 @@ def sign_function(transform_args, **kwargs):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_mqtt311_websocket_connect_basic_auth(self):
+    def test_mqtt311_websocket_connect_minimum(self):
+        test_retry_wrapper(self._test_mqtt311_websocket_connect_minimum)
+
+    def _test_mqtt311_websocket_connect_basic_auth(self):
         input_host_name = _get_env_variable("AWS_TEST_MQTT311_WS_MQTT_BASIC_AUTH_HOST")
         input_port = int(_get_env_variable("AWS_TEST_MQTT311_WS_MQTT_BASIC_AUTH_PORT"))
         input_username = _get_env_variable("AWS_TEST_MQTT311_BASIC_AUTH_USERNAME")
@@ -703,7 +764,10 @@ def sign_function(transform_args, **kwargs):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_mqtt311_websocket_connect_tls(self):
+    def test_mqtt311_websocket_connect_basic_auth(self):
+        test_retry_wrapper(self._test_mqtt311_websocket_connect_basic_auth)
+
+    def _test_mqtt311_websocket_connect_tls(self):
         input_host_name = _get_env_variable("AWS_TEST_MQTT311_WS_MQTT_TLS_HOST")
         input_port = int(_get_env_variable("AWS_TEST_MQTT311_WS_MQTT_TLS_PORT"))
 
@@ -727,7 +791,10 @@ def sign_function(transform_args, **kwargs):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
-    def test_mqtt311_websocket_connect_http_proxy_tls(self):
+    def test_mqtt311_websocket_connect_tls(self):
+        test_retry_wrapper(self._test_mqtt311_websocket_connect_tls)
+
+    def _test_mqtt311_websocket_connect_http_proxy_tls(self):
         input_proxy_host = _get_env_variable("AWS_TEST_MQTT311_PROXY_HOST")
         input_proxy_port = int(_get_env_variable("AWS_TEST_MQTT311_PROXY_PORT"))
         input_host_name = _get_env_variable("AWS_TEST_MQTT311_WS_MQTT_TLS_HOST")
@@ -761,6 +828,9 @@ def sign_function(transform_args, **kwargs):
         connection.connect().result(TIMEOUT)
         connection.disconnect().result(TIMEOUT)
 
+    def test_mqtt311_websocket_connect_http_proxy_tls(self):
+        test_retry_wrapper(self._test_mqtt311_websocket_connect_http_proxy_tls)
+
 
 if __name__ == 'main':
     unittest.main()