test_mqtt5to3_adapter.py

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0.

from concurrent.futures import Future
from awscrt import mqtt5, io, http, exceptions
from awscrt.mqtt import Connection, ConnectReturnCode, OnConnectionSuccessData, OnConnectionFailureData, OnConnectionClosedData, QoS
from test import NativeResourceTest
from test.test_mqtt5 import Mqtt5TestCallbacks, _get_env_variable, create_client_id
import unittest
import uuid


"""
# Environmental variables used by mqtt5 bindings tests

# AWS_TEST_MQTT5_DIRECT_MQTT_HOST - host to connect to in direct mqtt tests
# AWS_TEST_MQTT5_DIRECT_MQTT_PORT - port to connect to in direct mqtt tests
# AWS_TEST_MQTT5_DIRECT_MQTT_BASIC_AUTH_HOST - host to connect to in direct mqtt basic authentication tests
# AWS_TEST_MQTT5_DIRECT_MQTT_BASIC_AUTH_PORT - port to connect to in direct mqtt basic authentication tests
# AWS_TEST_MQTT5_DIRECT_MQTT_TLS_HOST - host to connect to in direct mqtt tls tests
# AWS_TEST_MQTT5_DIRECT_MQTT_TLS_PORT - port to connect to in direct mqtt tls tests
# AWS_TEST_MQTT5_WS_MQTT_HOST - host to connect to in websocket tests
# AWS_TEST_MQTT5_WS_MQTT_PORT - port to connect to in websocket tests
# AWS_TEST_MQTT5_WS_MQTT_BASIC_AUTH_HOST - host to connect to in websocket basic authentication tests
# AWS_TEST_MQTT5_WS_MQTT_BASIC_AUTH_PORT - port to connect to in websocket basic authentication tests
# AWS_TEST_MQTT5_WS_MQTT_TLS_HOST - host to connect to in websocket tls tests
# AWS_TEST_MQTT5_WS_MQTT_TLS_PORT - port to connect to in websocket tls tests
# AWS_TEST_MQTT5_IOT_CORE_HOST - host to connect to in MTLS tests
# AWS_TEST_MQTT5_BASIC_AUTH_USERNAME - username to use in basic authentication tests
# AWS_TEST_MQTT5_BASIC_AUTH_PASSWORD - password to use in basic authentication tests
# AWS_TEST_MQTT5_PROXY_HOST - name of http proxy host to use in proxy-based tests
# AWS_TEST_MQTT5_PROXY_PORT - port of http proxy host to use in proxy-based tests
# AWS_TEST_MQTT5_CERTIFICATE_FILE - certificate file path
# AWS_TEST_MQTT5_KEY_FILE - private key file path
# AWS_TEST_MQTT5_IOT_KEY_PATH - private key file path for MTLS tests
# AWS_TEST_MQTT5_IOT_CERTIFICATE_PATH - certificate file path for MTLS tests

"""

TIMEOUT = 100.0


class Mqtt311TestCallbacks():
    def __init__(self):
        self.last_exception = None

        self.future_connection_success = Future()
        self.future_connection_failure = Future()
        self.future_resumed = Future()
        self.future_interrupted = Future()
        self.future_closed = Future()
        self.future_message_received = Future()
        self.received_message = 0

    def on_connection_interrupted(self, connection: Connection, error: exceptions.AwsCrtError):
        if self.future_interrupted:
            self.future_interrupted.set_result(error)
            self.future_connection_success = self._reset_future(self.future_connection_success)

    def on_connection_resumed(self, connection: Connection, return_code: ConnectReturnCode, session_present: bool):
        if self.future_resumed:
            self.future_resumed.set_result({'return_code': return_code, "session_present": session_present})

    def on_connection_success(self, connection: Connection, callback_data: OnConnectionSuccessData):
        if self.future_connection_success:
            self.future_connection_success.set_result(
                {'return_code': callback_data.return_code, "session_present": callback_data.session_present})

    def on_connection_failure(self, connection: Connection, callback_data: OnConnectionFailureData):
        if self.future_connection_failure:
            self.future_connection_failure.set_result({'error': callback_data.error})

    def on_connection_closed(self, connection: Connection, callback_data: OnConnectionClosedData):
        if self.future_closed:
            self.future_closed.set_result({})

    def on_message(self, **kwargs):
        self.received_message += 1
        if self.future_message_received:
            self.future_message_received.set_result(kwargs)

    def _reset_future(self, future: Future):
        if future.done():
            return Future()
        return future


class Mqtt5to3AdapterTest(NativeResourceTest):
    TEST_MSG = 'NOTICE ME!'.encode('utf8')

    def _create_client(
            self,
            client_options: mqtt5.ClientOptions = None,
            callbacks: Mqtt5TestCallbacks = None):

        default_host = _get_env_variable("AWS_TEST_MQTT5_IOT_CORE_HOST")
        if client_options is None:
            client_options = mqtt5.ClientOptions(
                host_name=default_host,
                port=8883
            )
        if client_options.connect_options is None:
            client_options.connect_options = mqtt5.ConnectPacket()
            client_options.connect_options.client_id = create_client_id()

        if callbacks is not None:
            client_options.on_publish_callback_fn = callbacks.on_publish_received
            client_options.on_lifecycle_event_stopped_fn = callbacks.on_lifecycle_stopped
            client_options.on_lifecycle_event_attempting_connect_fn = callbacks.on_lifecycle_attempting_connect
            client_options.on_lifecycle_event_connection_success_fn = callbacks.on_lifecycle_connection_success
            client_options.on_lifecycle_event_connection_failure_fn = callbacks.on_lifecycle_connection_failure
            client_options.on_lifecycle_event_disconnection_fn = callbacks.on_lifecycle_disconnection

        client = mqtt5.Client(client_options)
        return client

    def _create_connection(self,
                           client: mqtt5.Client,
                           mqtt3_callbacks: Mqtt311TestCallbacks = None):
        connection = client.new_connection(on_connection_closed=mqtt3_callbacks.on_connection_closed,
                                           on_connection_failure=mqtt3_callbacks.on_connection_failure,
                                           on_connection_success=mqtt3_callbacks.on_connection_success,
                                           on_connection_resumed=mqtt3_callbacks.on_connection_resumed,
                                           on_connection_interrupted=mqtt3_callbacks.on_connection_interrupted,)
        return connection

    # # ==============================================================
    # #             MQTT5 CLIENT SETUP
    # # ==============================================================

    def _setup_direct_connect_minimum(self):
        input_host_name = _get_env_variable("AWS_TEST_MQTT5_DIRECT_MQTT_HOST")
        input_port = int(_get_env_variable("AWS_TEST_MQTT5_DIRECT_MQTT_PORT"))

        client_options = mqtt5.ClientOptions(
            host_name=input_host_name,
            port=input_port
        )
        callbacks = Mqtt5TestCallbacks()
        return self._create_client(client_options=client_options, callbacks=callbacks), callbacks

    def _setup_direct_connect_basic_auth(self):
        input_username = _get_env_variable("AWS_TEST_MQTT5_BASIC_AUTH_USERNAME")
        input_password = _get_env_variable("AWS_TEST_MQTT5_BASIC_AUTH_PASSWORD")
        input_host_name = _get_env_variable("AWS_TEST_MQTT5_DIRECT_MQTT_BASIC_AUTH_HOST")
        input_port = int(_get_env_variable("AWS_TEST_MQTT5_DIRECT_MQTT_BASIC_AUTH_PORT"))

        connect_options = mqtt5.ConnectPacket(
            client_id=create_client_id(),
            username=input_username,
            password=input_password
        )
        client_options = mqtt5.ClientOptions(
            host_name=input_host_name,
            port=input_port,
            connect_options=connect_options
        )
        callbacks = Mqtt5TestCallbacks()
        return self._create_client(client_options=client_options, callbacks=callbacks), callbacks

    def _setup_direct_connect_mutual_tls(self):
        input_host_name = _get_env_variable("AWS_TEST_MQTT5_IOT_CORE_HOST")
        input_cert = _get_env_variable("AWS_TEST_MQTT5_IOT_CORE_RSA_CERT")
        input_key = _get_env_variable("AWS_TEST_MQTT5_IOT_CORE_RSA_KEY")

        client_options = mqtt5.ClientOptions(
            host_name=input_host_name,
            port=8883
        )
        tls_ctx_options = io.TlsContextOptions.create_client_with_mtls_from_path(
            input_cert,
            input_key
        )
        client_options.tls_ctx = io.ClientTlsContext(tls_ctx_options)

        callbacks = Mqtt5TestCallbacks()
        return self._create_client(client_options=client_options, callbacks=callbacks), callbacks

    # ==============================================================
    #             WEBSOCKET CONNECT TEST SETUP
    # ==============================================================

    def _setup_websocket_connect_minimum(self):
        input_host_name = _get_env_variable("AWS_TEST_MQTT5_WS_MQTT_HOST")
        input_port = int(_get_env_variable("AWS_TEST_MQTT5_WS_MQTT_PORT"))

        client_options = mqtt5.ClientOptions(
            host_name=input_host_name,
            port=input_port
        )
        callbacks = Mqtt5TestCallbacks()
        client_options.websocket_handshake_transform = callbacks.ws_handshake_transform

        return self._create_client(client_options=client_options, callbacks=callbacks), callbacks

    def _setup_websocket_connect_http_proxy_tls(self):
        input_host_name = _get_env_variable("AWS_TEST_MQTT5_WS_MQTT_TLS_HOST")
        input_port = int(_get_env_variable("AWS_TEST_MQTT5_WS_MQTT_TLS_PORT"))
        input_proxy_host = _get_env_variable("AWS_TEST_MQTT5_PROXY_HOST")
        input_proxy_port = int(_get_env_variable("AWS_TEST_MQTT5_PROXY_PORT"))

        client_options = mqtt5.ClientOptions(
            host_name=input_host_name,
            port=input_port
        )
        tls_ctx_options = io.TlsContextOptions()
        tls_ctx_options.verify_peer = False
        client_options.tls_ctx = io.ClientTlsContext(tls_ctx_options)

        http_proxy_options = http.HttpProxyOptions(
            host_name=input_proxy_host,
            port=input_proxy_port
        )
        http_proxy_options.connection_type = http.HttpProxyConnectionType.Tunneling
        http_proxy_options.auth_type = http.HttpProxyAuthenticationType.Nothing
        client_options.http_proxy_options = http_proxy_options

        callbacks = Mqtt5TestCallbacks()
        client_options.websocket_handshake_transform = callbacks.ws_handshake_transform

        return self._create_client(client_options=client_options, callbacks=callbacks), callbacks

    def _test_with_mqtt3_connect(self, setup_client: callable):
        client, callbacks = setup_client()
        connection = client.new_connection()
        connection.connect().result(TIMEOUT)
        connection.disconnect().result(TIMEOUT)

    def _test_with_mqtt5_connect(self, setup_client: callable):
        client, callbacks = setup_client()
        connection = client.new_connection()
        client.start()
        callbacks.future_connection_success.result(TIMEOUT)
        client.stop()
        callbacks.future_stopped.result(TIMEOUT)

    # ==============================================================
    #             CONNECTION TEST HELPER FUNCTIONS
    # ==============================================================

    def _test_with_mqtt3_connect(self, setup_client: callable):
        client, callbacks = setup_client()
        connection = client.new_connection()
        connection.connect().result(TIMEOUT)
        connection.disconnect().result(TIMEOUT)

    def _test_with_mqtt5_connect(self, setup_client: callable):
        client, callbacks = setup_client()
        connection = client.new_connection()
        client.start()
        callbacks.future_connection_success.result(TIMEOUT)
        client.stop()
        callbacks.future_stopped.result(TIMEOUT)

    # ==============================================================
    #             CREATION TEST CASES
    # ==============================================================

    def test_client_creation_minimum(self):
        client = self._create_client()
        connection = client.new_connection()

    def test_client_creation_maximum(self):
        input_host_name = _get_env_variable("AWS_TEST_MQTT5_IOT_CORE_HOST")

        user_properties = []
        user_properties.append(mqtt5.UserProperty(name="name1", value="value1"))
        user_properties.append(mqtt5.UserProperty(name="name2", value="value2"))

        publish_packet = mqtt5.PublishPacket(
            payload="TEST_PAYLOAD",
            qos=mqtt5.QoS.AT_LEAST_ONCE,
            retain=False,
            topic="TEST_TOPIC",
            payload_format_indicator=mqtt5.PayloadFormatIndicator.AWS_MQTT5_PFI_UTF8,
            message_expiry_interval_sec=10,
            topic_alias=1,
            response_topic="TEST_RESPONSE_TOPIC",
            correlation_data="TEST_CORRELATION_DATA",
            content_type="TEST_CONTENT_TYPE",
            user_properties=user_properties
        )

        connect_options = mqtt5.ConnectPacket(
            keep_alive_interval_sec=10,
            client_id=create_client_id(),
            username="USERNAME",
            password="PASSWORD",
            session_expiry_interval_sec=100,
            request_response_information=1,
            request_problem_information=1,
            receive_maximum=1000,
            maximum_packet_size=10000,
            will_delay_interval_sec=1000,
            will=publish_packet,
            user_properties=user_properties
        )
        client_options = mqtt5.ClientOptions(
            host_name=input_host_name,
            port=8883,
            connect_options=connect_options,
            session_behavior=mqtt5.ClientSessionBehaviorType.CLEAN,
            extended_validation_and_flow_control_options=mqtt5.ExtendedValidationAndFlowControlOptions.AWS_IOT_CORE_DEFAULTS,
            offline_queue_behavior=mqtt5.ClientOperationQueueBehaviorType.FAIL_ALL_ON_DISCONNECT,
            retry_jitter_mode=mqtt5.ExponentialBackoffJitterMode.DECORRELATED,
            min_reconnect_delay_ms=100,
            max_reconnect_delay_ms=50000,
            min_connected_time_to_reset_reconnect_delay_ms=1000,
            ping_timeout_ms=1000,
            connack_timeout_ms=1000,
            ack_timeout_sec=100)
        client = self._create_client(client_options=client_options)
        connection = client.new_connection()

    # ==============================================================
    #         CONNECT THROUGH MQTT311 INTERFACE TEST CASES
    # ==============================================================
    def test_direct_connect_through_mqtt311_minimum(self):
        self._test_with_mqtt3_connect(self._setup_direct_connect_minimum)

    # def test_direct_connect_through_mqtt311_basic_auth(self):
    #     self._test_with_mqtt3_connect(self._setup_direct_connect_basic_auth)

    def test_direct_connect_through_mqtt311_mutual_tls(self):
        self._test_with_mqtt3_connect(self._setup_direct_connect_mutual_tls)

    # def test_direct_connect_through_mqtt311_websocket_minimum(self):
    #     self._test_with_mqtt3_connect(self._setup_websocket_connect_minimum)

    # def test_direct_connect_through_mqtt311_websocket_http_proxy_tls(self):
    #     self._test_with_mqtt3_connect(self._setup_websocket_connect_http_proxy_tls)

    # ==============================================================
    #         CONNECT THROUGH MQTT5 INTERFACE TEST CASES
    # ==============================================================

    def test_direct_connect_through_mqtt5_minimum(self):
        self._test_with_mqtt5_connect(self._setup_direct_connect_minimum)

    # def test_direct_connect_through_mqtt5_basic_auth(self):
    #     self._test_with_mqtt5_connect(self._setup_direct_connect_basic_auth)

    def test_direct_connect_through_mqtt5_mutual_tls(self):
        self._test_with_mqtt5_connect(self._setup_direct_connect_mutual_tls)

    # def test_direct_connect_through_mqtt5_websocket_minimum(self):
    #     self._test_with_mqtt5_connect(self._setup_websocket_connect_minimum)

    # def test_direct_connect_through_mqtt5_websocket_http_proxy_tls(self):
    #     self._test_with_mqtt5_connect(self._setup_websocket_connect_http_proxy_tls)

    # ==============================================================
    #             OPERATION TEST CASES
    # ==============================================================

    def test_operation_sub_unsub(self):
        TEST_TOPIC = '/test/topic/adapter' + str(uuid.uuid4())

        client, mqtt5_callbacks = self._setup_direct_connect_mutual_tls()
        mqtt311_callbacks = Mqtt311TestCallbacks()
        connection = self._create_connection(client, mqtt311_callbacks)

        connection.connect().result(TIMEOUT)

        # subscribe
        subscribed, packet_id = connection.subscribe(TEST_TOPIC, QoS.AT_LEAST_ONCE, mqtt311_callbacks.on_message)
        suback = subscribed.result(TIMEOUT)
        self.assertEqual(packet_id, suback['packet_id'])
        self.assertEqual(TEST_TOPIC, suback['topic'])
        self.assertIs(QoS.AT_LEAST_ONCE, suback['qos'])

        # publish
        published, packet_id = connection.publish(TEST_TOPIC, self.TEST_MSG, QoS.AT_LEAST_ONCE)
        puback = published.result(TIMEOUT)
        self.assertEqual(packet_id, puback['packet_id'])

        # receive message
        rcv = mqtt311_callbacks.future_message_received.result(TIMEOUT)
        self.assertEqual(TEST_TOPIC, rcv['topic'])
        self.assertEqual(self.TEST_MSG, rcv['payload'])
        self.assertFalse(rcv['dup'])
        self.assertEqual(QoS.AT_LEAST_ONCE, rcv['qos'])
        self.assertFalse(rcv['retain'])

        # unsubscribe
        unsubscribed, packet_id = connection.unsubscribe(TEST_TOPIC)
        unsuback = unsubscribed.result(TIMEOUT)
        self.assertEqual(packet_id, unsuback['packet_id'])

        # publish
        published, packet_id = connection.publish(TEST_TOPIC, self.TEST_MSG, QoS.AT_LEAST_ONCE)
        puback = published.result(TIMEOUT)
        self.assertEqual(packet_id, puback['packet_id'])

        self.assertEqual(mqtt311_callbacks.received_message, 1)
        self.assertEqual(mqtt5_callbacks.on_publish_received_counter, 1)

        connection.disconnect().result(TIMEOUT)

    def test_operation_null_ack(self):
        TEST_TOPIC = '/test/topic/adapter' + str(uuid.uuid4())
        exception_occurred = False

        client, mqtt5_callbacks = self._setup_direct_connect_mutual_tls()
        mqtt311_callbacks = Mqtt311TestCallbacks()
        connection = self._create_connection(client, mqtt311_callbacks)

        # publish an offline packet so that the publish operation would be incomplete
        published, packet_id = connection.publish(TEST_TOPIC, self.TEST_MSG, QoS.AT_LEAST_ONCE)

        # Manually destroyed the connection so that the incomplete publish operation should fail with exception
        del connection
        try:
            published.result(TIMEOUT)
        except TimeoutError:
            # Directly failed the test if the result time out
            self.assertTrue(False, "Operation Time out. The connection does not fail the callback on destroy. ")
        except Exception:
            exception_occurred = True

        assert (exception_occurred)

    # ==============================================================
    #                 MQTT311 CALLBACK TEST CASES
    # ==============================================================
    def test_connection_success_callback(self):
        client, _ = self._setup_direct_connect_minimum()
        mqtt311_callbacks = Mqtt311TestCallbacks()
        connection = self._create_connection(client, mqtt311_callbacks)

        connection.connect().result(TIMEOUT)
        mqtt311_callbacks.future_connection_success.result(TIMEOUT)
        connection.disconnect().result(TIMEOUT)

    def test_connection_failure_callback(self):
        client_options = mqtt5.ClientOptions(
            host_name="badhost",
            port=1883
        )
        callbacks = Mqtt5TestCallbacks()
        client = self._create_client(client_options=client_options, callbacks=callbacks)
        mqtt311_callbacks = Mqtt311TestCallbacks()
        connection = self._create_connection(client, mqtt311_callbacks)

        try:
            connection.connect().result(TIMEOUT)
        except Exception:
            exception_occurred = True
        self.assertTrue(exception_occurred, "Exception did not occur when connecting with invalid arguments!")

        failure_data = mqtt311_callbacks.future_connection_failure.result(TIMEOUT)
        self.assertTrue(failure_data['error'] is not None)

    def test_connection_interrupted_and_resumed_callback(self):
        input_host_name = _get_env_variable("AWS_TEST_MQTT5_DIRECT_MQTT_HOST")
        input_port = int(_get_env_variable("AWS_TEST_MQTT5_DIRECT_MQTT_PORT"))

        client_options = mqtt5.ClientOptions(
            host_name=input_host_name,
            port=input_port
        )
        client_options.connect_options = mqtt5.ConnectPacket()
        client_options.connect_options.client_id = create_client_id()

        client1 = self._create_client(client_options=client_options)
        client2 = self._create_client(client_options=client_options)

        mqtt311_callbacks1 = Mqtt311TestCallbacks()
        mqtt311_callbacks2 = Mqtt311TestCallbacks()

        # Create two connection with the same client id
        connection1 = self._create_connection(client1, mqtt311_callbacks1)
        connection2 = self._create_connection(client2, mqtt311_callbacks2)

        connection1.connect().result(TIMEOUT)

        connection2.connect().result(TIMEOUT)

        # with connection2 connected, connection1 should be interrupted
        mqtt311_callbacks1.future_interrupted.result(TIMEOUT)
        # The connection1 should automatically re-connected
        mqtt311_callbacks1.future_resumed.result(TIMEOUT)

        # At this point, connection2 should be interrupted
        mqtt311_callbacks2.future_interrupted.result(TIMEOUT)

        connection1.disconnect().result(TIMEOUT)
        connection2.disconnect().result(TIMEOUT)

    # ==============================================================
    #                 ADAPTER TEST CASES
    # ==============================================================
    def test_multiple_adapters(self):
        TEST_TOPIC1 = '/test/topic/adapter1' + str(uuid.uuid4())
        TEST_TOPIC2 = '/test/topic/adapter2' + str(uuid.uuid4())
        TEST_TOPIC3 = '/test/topic/adapter3' + str(uuid.uuid4())

        client, mqtt5_callbacks = self._setup_direct_connect_mutual_tls()
        mqtt311_callbacks1 = Mqtt311TestCallbacks()
        connection1 = self._create_connection(client, mqtt3_callbacks=mqtt311_callbacks1)
        mqtt311_callbacks2 = Mqtt311TestCallbacks()
        connection2 = self._create_connection(client, mqtt3_callbacks=mqtt311_callbacks2)
        mqtt311_callbacks3 = Mqtt311TestCallbacks()
        connection3 = self._create_connection(client, mqtt3_callbacks=mqtt311_callbacks3)

        client.start()
        mqtt5_callbacks.future_connection_success.result(TIMEOUT)

        # subscribe
        subscribed, packet_id = connection1.subscribe(
            TEST_TOPIC1, QoS.AT_LEAST_ONCE, mqtt311_callbacks1.on_message)
        suback = subscribed.result(TIMEOUT)
        subscribed, packet_id = connection2.subscribe(
            TEST_TOPIC2, QoS.AT_LEAST_ONCE, mqtt311_callbacks2.on_message)
        suback = subscribed.result(TIMEOUT)
        subscribed, packet_id = connection3.subscribe(
            TEST_TOPIC3, QoS.AT_LEAST_ONCE, mqtt311_callbacks3.on_message)
        suback = subscribed.result(TIMEOUT)

        # publish on topic1
        publish_packet = mqtt5.PublishPacket(
            payload=self.TEST_MSG,
            topic=TEST_TOPIC1,
            qos=mqtt5.QoS.AT_LEAST_ONCE)

        client.publish(publish_packet=publish_packet).result(TIMEOUT)

        # only connection1 should receive message
        mqtt311_callbacks1.future_message_received.result(TIMEOUT)

        self.assertEqual(mqtt311_callbacks1.received_message, 1)
        self.assertEqual(mqtt311_callbacks2.received_message, 0)
        self.assertEqual(mqtt311_callbacks3.received_message, 0)

        # publish on topic2
        publish_packet = mqtt5.PublishPacket(
            payload=self.TEST_MSG,
            topic=TEST_TOPIC2,
            qos=mqtt5.QoS.AT_LEAST_ONCE)

        client.publish(publish_packet=publish_packet).result(TIMEOUT)

        # connection2 should receive message
        mqtt311_callbacks2.future_message_received.result(TIMEOUT)

        self.assertEqual(mqtt311_callbacks1.received_message, 1)
        self.assertEqual(mqtt311_callbacks2.received_message, 1)
        self.assertEqual(mqtt311_callbacks3.received_message, 0)

        # publish on topic3
        publish_packet = mqtt5.PublishPacket(
            payload=self.TEST_MSG,
            topic=TEST_TOPIC3,
            qos=mqtt5.QoS.AT_LEAST_ONCE)

        client.publish(publish_packet=publish_packet).result(TIMEOUT)

        # connection3 should receive message
        mqtt311_callbacks3.future_message_received.result(TIMEOUT)
        self.assertEqual(mqtt311_callbacks1.received_message, 1)
        self.assertEqual(mqtt311_callbacks2.received_message, 1)
        self.assertEqual(mqtt311_callbacks3.received_message, 1)

        client.stop()
        mqtt5_callbacks.future_stopped.result(TIMEOUT)


if __name__ == 'main':
    unittest.main()