# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0. from awscrt.mqtt_request_response import StreamingOperationOptions from test import test_retry_wrapper, NativeResourceTest from awscrt import io, mqtt5, mqtt_request_response, mqtt from concurrent.futures import Future import os import time import unittest import uuid TIMEOUT = 30.0 def create_client_id(): return f"aws-crt-python-unit-test-{uuid.uuid4()}" def _get_env_variable(env_name): env_data = os.environ.get(env_name) if not env_data: raise unittest.SkipTest(f"test requires env var: {env_name}") return env_data class MqttRequestResponse5TestCallbacks(): def __init__(self): self.future_connection_success = Future() self.future_stopped = Future() def ws_handshake_transform(self, transform_args): transform_args.set_done() def on_publish_received(self, publish_received_data: mqtt5.PublishReceivedData): pass def on_lifecycle_stopped(self, lifecycle_stopped: mqtt5.LifecycleStoppedData): if self.future_stopped: self.future_stopped.set_result(None) def on_lifecycle_attempting_connect(self, lifecycle_attempting_connect: mqtt5.LifecycleAttemptingConnectData): pass def on_lifecycle_connection_success(self, lifecycle_connection_success: mqtt5.LifecycleConnectSuccessData): if self.future_connection_success: if self.future_connection_success.done(): pass else: self.future_connection_success.set_result(lifecycle_connection_success) def on_lifecycle_connection_failure(self, lifecycle_connection_failure: mqtt5.LifecycleConnectFailureData): if self.future_connection_success: if self.future_connection_success.done(): pass else: self.future_connection_success.set_exception(lifecycle_connection_failure.exception) def on_lifecycle_disconnection(self, lifecycle_disconnect_data: mqtt5.LifecycleDisconnectData): pass def _empty_response_paths(options): options.response_paths = [] def _invalidate_response_path_topic(options): options.response_paths[0].topic = "a/#/b" def _none_response_path_topic(options): options.response_paths[0].topic = None def _missing_response_path_topic(options): del options.response_paths[0].topic def _type_mismatch_response_path_topic(options): options.response_paths[0].topic = 57.3 def _type_mismatch_response_path_correlation_token_json_path(options): options.response_paths[0].correlation_token_json_path = [] def _type_mismatch_response_paths(options): options.response_paths = "hello" def _invalidate_subscription_topic_filter(options): options.subscription_topic_filters[0] = "a/#/c" def _type_mismatch_subscription_topic_filter(options): options.subscription_topic_filters[0] = ["thirty", 30] def _type_mismatch_subscriptions(options): options.subscription_topic_filters = 50 def _empty_subscription_topic_filters(options): options.subscription_topic_filters = [] def _none_publish_topic(options): options.publish_topic = None def _bad_publish_topic(options): options.publish_topic = "#/b/c" def _type_mismatch_publish_topic(options): options.publish_topic = [["oof"]] def _type_mismatch_correlation_token(options): options.correlation_token = [-1] def _subscription_topic_filter_none(options): options.subscription_topic_filter = None def _type_mismatch_stream_subscription_topic_filter(options): options.subscription_topic_filter = 6 def _type_mismatch_subscription_status_listener(options): options.subscription_status_listener = "string" def _type_mismatch_incoming_publish_listener(options): options.incoming_publish_listener = "string" class MqttRequestResponseClientTest(NativeResourceTest): def _create_client5(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) client_options.connect_options = mqtt5.ConnectPacket() client_options.connect_options.client_id = create_client_id() callbacks = MqttRequestResponse5TestCallbacks() client_options.on_lifecycle_event_stopped_fn = callbacks.on_lifecycle_stopped 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_stopped_fn = callbacks.on_lifecycle_stopped protocol_client = mqtt5.Client(client_options) protocol_client.start() callbacks.future_connection_success.result(TIMEOUT) return protocol_client, callbacks def _shutdown5(self, protocol_client, callbacks): protocol_client.stop() callbacks.future_stopped.result(TIMEOUT) def _create_client311(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") tls_ctx_options = io.TlsContextOptions.create_client_with_mtls_from_path( input_cert, input_key ) tls_ctx = io.ClientTlsContext(tls_ctx_options) client = mqtt.Client(None, tls_ctx) protocol_client = mqtt.Connection( client=client, client_id=create_client_id(), host_name=input_host_name, port=8883, ping_timeout_ms=10000, keep_alive_secs=30 ) protocol_client.connect().result(TIMEOUT) return protocol_client def _shutdown311(self, protocol_client): protocol_client.disconnect().result(TIMEOUT) def _create_rr_client( self, protocol_client, max_request_response_subscriptions, max_streaming_subscriptions, operation_timeout_seconds): rr_client_options = mqtt_request_response.ClientOptions( max_request_response_subscriptions, max_streaming_subscriptions) rr_client_options.operation_timeout_in_seconds = operation_timeout_seconds rr_client = mqtt_request_response.Client(protocol_client, rr_client_options) return rr_client def _do_mqtt5_test(self, test_callable): (protocol_client, callbacks) = self._create_client5() test_callable(protocol_client) self._shutdown5(protocol_client, callbacks) def _do_mqtt311_test(self, test_callable): protocol_client = self._create_client311() test_callable(protocol_client) self._shutdown311(protocol_client) def _create_get_shadow_request(self, thing_name, use_correlation_token): topic_prefix = f"$aws/things/{thing_name}/shadow/get" request_options = mqtt_request_response.RequestOptions( subscription_topic_filters=[f"{topic_prefix}/+"], response_paths=[ mqtt_request_response.ResponsePath(topic=f"{topic_prefix}/accepted"), mqtt_request_response.ResponsePath(topic=f"{topic_prefix}/rejected") ], publish_topic=topic_prefix, payload="{}".encode() ) if use_correlation_token: correlation_token = f"{uuid.uuid4()}" request_options.response_paths[0].correlation_token_json_path = "clientToken" request_options.response_paths[1].correlation_token_json_path = "clientToken" request_options.payload = f'{{"clientToken":"{correlation_token}"}}'.encode() request_options.correlation_token = correlation_token return request_options def _do_get_shadow_success_no_such_shadow(self, rr_client, thing_name, use_correlation_token): request_options = self._create_get_shadow_request(thing_name, use_correlation_token) request_future = rr_client.make_request(request_options) response = request_future.result() self.assertEqual(request_options.response_paths[1].topic, response.topic, "Expected response to come in on rejected topic") payload = str(response.payload) self.assertIn("No shadow exists with name", payload) def _do_get_shadow_success_no_such_shadow_test(self, protocol_client, use_correlation_token): rr_client = self._create_rr_client(protocol_client, 2, 2, 30) thing_name = f"tn-{uuid.uuid4()}" self._do_get_shadow_success_no_such_shadow(rr_client, thing_name, use_correlation_token) def _do_get_shadow_success(self, rr_client, thing_name, use_correlation_token): request_options = self._create_get_shadow_request(thing_name, use_correlation_token) request_future = rr_client.make_request(request_options) response = request_future.result() self.assertEqual(request_options.response_paths[0].topic, response.topic, "Expected response to come in on accepted topic") payload = str(response.payload) self.assertIn("magic", payload) def _do_update_shadow_success(self, rr_client, thing_name, use_correlation_token): topic_prefix = f"$aws/things/{thing_name}/shadow/update" request_options = mqtt_request_response.RequestOptions( subscription_topic_filters=[f"{topic_prefix}/accepted", f"{topic_prefix}/rejected"], response_paths=[ mqtt_request_response.ResponsePath(topic=f"{topic_prefix}/accepted"), mqtt_request_response.ResponsePath(topic=f"{topic_prefix}/rejected") ], publish_topic=topic_prefix, payload="".encode() ) desired_state = f'{{"magic":"value"}}' if use_correlation_token: correlation_token = f"{uuid.uuid4()}" payload = f'{{"clientToken": "{correlation_token}","state": {{"desired": {desired_state}}}}}' request_options.response_paths[0].correlation_token_json_path = "clientToken" request_options.response_paths[1].correlation_token_json_path = "clientToken" request_options.payload = payload.encode() request_options.correlation_token = correlation_token else: request_options.payload = f'{{"state":{{"desired":{desired_state}}}}}'.encode() request_future = rr_client.make_request(request_options) response = request_future.result() self.assertEqual(request_options.response_paths[0].topic, response.topic, "Expected response to come in on accepted topic") payload = str(response.payload) self.assertIn("magic", payload) def _do_delete_shadow_success(self, rr_client, thing_name, use_correlation_token): topic_prefix = f"$aws/things/{thing_name}/shadow/delete" request_options = mqtt_request_response.RequestOptions( subscription_topic_filters=[f"{topic_prefix}/+"], response_paths=[ mqtt_request_response.ResponsePath(topic=f"{topic_prefix}/accepted"), mqtt_request_response.ResponsePath(topic=f"{topic_prefix}/rejected") ], publish_topic=topic_prefix, payload="{}".encode() ) if use_correlation_token: correlation_token = f"{uuid.uuid4()}" request_options.response_paths[0].correlation_token_json_path = "clientToken" request_options.response_paths[1].correlation_token_json_path = "clientToken" request_options.payload = f'{{"clientToken":"{correlation_token}"}}'.encode() request_options.correlation_token = correlation_token request_future = rr_client.make_request(request_options) response = request_future.result() self.assertEqual(request_options.response_paths[0].topic, response.topic, "Expected response to come in on accepted topic") payload = str(response.payload) self.assertIn("version", payload) def _do_update_delete_shadow_success_test(self, protocol_client, use_correlation_token): rr_client = self._create_rr_client(protocol_client, 2, 2, 30) # get should return non-existence thing_name = f"tn-{uuid.uuid4()}" try: self._do_get_shadow_success_no_such_shadow(rr_client, thing_name, use_correlation_token) # update shadow to create self._do_update_shadow_success(rr_client, thing_name, use_correlation_token) # eventual consistency worries time.sleep(2) # get should now return the shadow state self._do_get_shadow_success(rr_client, thing_name, use_correlation_token) finally: # delete shadow self._do_delete_shadow_success(rr_client, thing_name, use_correlation_token) def _do_stream_success_test(self, protocol_client): rr_client = self._create_rr_client(protocol_client, 2, 2, 30) thing_name = f"tn-{uuid.uuid4()}" topic = f"not/a/valid/topic/{thing_name}" payload = b'hello' subscribed_future = Future() publish_received_future = Future() def on_subscription_status_changed(event): if event.type == mqtt_request_response.SubscriptionStatusEventType.SUBSCRIPTION_ESTABLISHED: subscribed_future.set_result(True) def on_incoming_publish(event): if event.topic == topic and event.payload == payload: publish_received_future.set_result(True) stream_options = StreamingOperationOptions(topic, on_subscription_status_changed, on_incoming_publish) stream = rr_client.create_stream(stream_options) stream.open() assert subscribed_future.result(30) if isinstance(protocol_client, mqtt5.Client): publish = mqtt5.PublishPacket( payload=payload, qos=mqtt5.QoS.AT_LEAST_ONCE, topic=topic ) protocol_client.publish(publish) else: protocol_client.publish(topic, payload, mqtt.QoS.AT_LEAST_ONCE) assert publish_received_future.result(30) def _do_get_shadow_failure_test(self, protocol_client, options_transform): rr_client = self._create_rr_client(protocol_client, 2, 2, 30) thing_name = f"tn-{uuid.uuid4()}" request_options = self._create_get_shadow_request(thing_name, True) options_transform(request_options) self.assertRaises(Exception, lambda: rr_client.make_request(request_options)) def _do_get_shadow_future_failure_test(self, protocol_client, options_transform): rr_client = self._create_rr_client(protocol_client, 2, 2, 30) thing_name = f"tn-{uuid.uuid4()}" request_options = self._create_get_shadow_request(thing_name, True) options_transform(request_options) response_future = rr_client.make_request(request_options) self.assertRaises(Exception, lambda: response_future.result()) def _do_create_stream_failure_test(self, protocol_client, options_transform): rr_client = self._create_rr_client(protocol_client, 2, 2, 30) thing_name = f"tn-{uuid.uuid4()}" stream_options = mqtt_request_response.StreamingOperationOptions(f"not/a/real/thing/{thing_name}") options_transform(stream_options) self.assertRaises(Exception, lambda: rr_client.create_stream(stream_options)) # ============================================================== # CREATION SUCCESS TEST CASES # ============================================================== def _test_client_creation_success5(self): self._do_mqtt5_test(lambda protocol_client: self._create_rr_client(protocol_client, 2, 2, 30)) def test_client_creation_success5(self): test_retry_wrapper(self._test_client_creation_success5) def _test_client_creation_success311(self): self._do_mqtt311_test(lambda protocol_client: self._create_rr_client(protocol_client, 2, 2, 30)) def test_client_creation_success311(self): test_retry_wrapper(self._test_client_creation_success311) # ============================================================== # CREATION FAILURE TEST CASES # ============================================================== def test_client_creation_failure_no_protocol_client(self): self.assertRaises(Exception, self._create_rr_client, None, 2, 2, 30) def test_client_creation_failure_zero_request_response_subscriptions5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 0, 2, 30)) def test_client_creation_failure_zero_request_response_subscriptions311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 0, 2, 30)) def test_client_creation_failure_negative_request_response_subscriptions5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, -2, 2, 30)) def test_client_creation_failure_negative_request_response_subscriptions311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, -2, 2, 30)) def test_client_creation_failure_no_request_response_subscriptions5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, None, 2, 30)) def test_client_creation_failure_no_request_response_subscriptions311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, None, 2, 30)) def test_client_creation_failure_request_response_subscriptions_type_mismatch5(self): self._do_mqtt5_test( lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, "None", 2, 30)) def test_client_creation_failure_request_response_subscriptions_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, "None", 2, 30)) def test_client_creation_failure_negative_streaming_subscriptions5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, -2, 30)) def test_client_creation_failure_negative_streaming_subscriptions311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, -2, 30)) def test_client_creation_failure_no_streaming_subscriptions5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, None, 30)) def test_client_creation_failure_no_streaming_subscriptions311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, None, 30)) def test_client_creation_failure_streaming_subscriptions_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, [], 30)) def test_client_creation_failure_streaming_subscriptions_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, {}, 30)) def test_client_creation_failure_negative_operation_timeout5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, 2, -30)) def test_client_creation_failure_negative_operation_timeout311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, 2, -30)) def test_client_creation_failure_no_operation_timeout5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, 2, None)) def test_client_creation_failure_no_operation_timeout311(self): self._do_mqtt311_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, 2, None)) def test_client_creation_failure_operation_timeout_invalid5(self): self._do_mqtt5_test(lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, 2, ["uffdah"])) def test_client_creation_failure_operation_timeout_invalid311(self): self._do_mqtt311_test( lambda protocol_client: self.assertRaises( Exception, self._create_rr_client, protocol_client, 2, 2, 777777777777777777777777777777777777)) # ============================================================== # make_request SUCCESS TEST CASES # ============================================================== def _test_get_shadow_success_no_such_shadow5(self): self._do_mqtt5_test( lambda protocol_client: self._do_get_shadow_success_no_such_shadow_test( protocol_client, True)) def test_get_shadow_success_no_such_shadow5(self): test_retry_wrapper(self._test_get_shadow_success_no_such_shadow5) def _test_get_shadow_success_no_such_shadow311(self): self._do_mqtt311_test( lambda protocol_client: self._do_get_shadow_success_no_such_shadow_test( protocol_client, True)) def test_get_shadow_success_no_such_shadow311(self): test_retry_wrapper(self._test_get_shadow_success_no_such_shadow311) def _test_get_shadow_success_no_such_shadow_no_correlation_token5(self): self._do_mqtt5_test( lambda protocol_client: self._do_get_shadow_success_no_such_shadow_test( protocol_client, False)) def test_get_shadow_success_no_such_shadow_no_correlation_token5(self): test_retry_wrapper(self._test_get_shadow_success_no_such_shadow_no_correlation_token5) def _test_get_shadow_success_no_such_shadow_no_correlation_token311(self): self._do_mqtt311_test( lambda protocol_client: self._do_get_shadow_success_no_such_shadow_test( protocol_client, False)) def test_get_shadow_success_no_such_shadow_no_correlation_token311(self): test_retry_wrapper(self._test_get_shadow_success_no_such_shadow_no_correlation_token311) def _test_update_delete_shadow_success5(self): self._do_mqtt5_test(lambda protocol_client: self._do_update_delete_shadow_success_test(protocol_client, True)) def test_update_delete_shadow_success5(self): test_retry_wrapper(self._test_update_delete_shadow_success5) def _test_update_delete_shadow_success311(self): self._do_mqtt311_test(lambda protocol_client: self._do_update_delete_shadow_success_test(protocol_client, True)) def test_update_delete_shadow_success311(self): test_retry_wrapper(self._test_update_delete_shadow_success311) def _test_update_delete_shadow_success_no_correlation_token5(self): self._do_mqtt5_test(lambda protocol_client: self._do_update_delete_shadow_success_test(protocol_client, False)) def test_update_delete_shadow_success_no_correlation_token5(self): test_retry_wrapper(self._test_update_delete_shadow_success_no_correlation_token5) def _test_update_delete_shadow_success_no_correlation_token311(self): self._do_mqtt311_test( lambda protocol_client: self._do_update_delete_shadow_success_test( protocol_client, False)) def test_update_delete_shadow_success_no_correlation_token311(self): test_retry_wrapper(self._test_update_delete_shadow_success_no_correlation_token311) # ============================================================== # make_request FAILURE TEST CASES # ============================================================== def _test_get_shadow_failure_no_response_paths5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _empty_response_paths(options))) def test_get_shadow_failure_no_response_paths5(self): test_retry_wrapper(self._test_get_shadow_failure_no_response_paths5) def _test_get_shadow_failure_no_response_paths311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _empty_response_paths(options))) def test_get_shadow_failure_no_response_paths311(self): test_retry_wrapper(self._test_get_shadow_failure_no_response_paths311) def _test_get_shadow_failure_invalid_response_path_topic5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_future_failure_test( protocol_client, lambda options: _invalidate_response_path_topic(options))) def test_get_shadow_failure_invalid_response_path_topic5(self): test_retry_wrapper(self._test_get_shadow_failure_invalid_response_path_topic5) def _test_get_shadow_failure_invalid_response_path_topic311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_future_failure_test( protocol_client, lambda options: _invalidate_response_path_topic(options))) def test_get_shadow_failure_invalid_response_path_topic311(self): test_retry_wrapper(self._test_get_shadow_failure_invalid_response_path_topic311) def _test_get_shadow_failure_none_response_path_topic5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _none_response_path_topic(options))) def test_get_shadow_failure_none_response_path_topic5(self): test_retry_wrapper(self._test_get_shadow_failure_none_response_path_topic5) def _test_get_shadow_failure_none_response_path_topic311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _none_response_path_topic(options))) def test_get_shadow_failure_none_response_path_topic311(self): test_retry_wrapper(self._test_get_shadow_failure_none_response_path_topic311) def _test_get_shadow_failure_missing_response_path_topic5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _missing_response_path_topic(options))) def test_get_shadow_failure_missing_response_path_topic5(self): test_retry_wrapper(self._test_get_shadow_failure_missing_response_path_topic5) def _test_get_shadow_failure_missing_response_path_topic311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _missing_response_path_topic(options))) def test_get_shadow_failure_missing_response_path_topic311(self): test_retry_wrapper(self._test_get_shadow_failure_missing_response_path_topic311) def _test_get_shadow_failure_response_path_topic_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_response_path_topic(options))) def test_get_shadow_failure_response_path_topic_type_mismatch5(self): test_retry_wrapper(self._test_get_shadow_failure_response_path_topic_type_mismatch5) def _test_get_shadow_failure_response_path_topic_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_response_path_topic(options))) def test_get_shadow_failure_response_path_topic_type_mismatch311(self): test_retry_wrapper(self._test_get_shadow_failure_response_path_topic_type_mismatch311) def _test_get_shadow_failure_response_path_correlation_token_json_path_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_response_path_correlation_token_json_path(options))) def test_get_shadow_failure_response_path_correlation_token_json_path_type_mismatch5(self): test_retry_wrapper(self._test_get_shadow_failure_response_path_correlation_token_json_path_type_mismatch5) def _test_get_shadow_failure_response_path_correlation_token_json_path_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_response_path_correlation_token_json_path(options))) def test_get_shadow_failure_response_path_correlation_token_json_path_type_mismatch311(self): test_retry_wrapper(self._test_get_shadow_failure_response_path_correlation_token_json_path_type_mismatch311) def _test_get_shadow_failure_response_paths_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_response_paths(options))) def test_get_shadow_failure_response_paths_type_mismatch5(self): test_retry_wrapper(self._test_get_shadow_failure_response_paths_type_mismatch5) def _test_get_shadow_failure_response_paths_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_response_paths(options))) def test_get_shadow_failure_response_paths_type_mismatch311(self): test_retry_wrapper(self._test_get_shadow_failure_response_paths_type_mismatch311) def _test_get_shadow_failure_invalid_subscription_topic5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_future_failure_test( protocol_client, lambda options: _invalidate_subscription_topic_filter(options))) def test_get_shadow_failure_invalid_subscription_topic5(self): test_retry_wrapper(self._test_get_shadow_failure_invalid_subscription_topic5) def _test_get_shadow_failure_invalid_subscription_topic311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_future_failure_test( protocol_client, lambda options: _invalidate_subscription_topic_filter(options))) def test_get_shadow_failure_invalid_subscription_topic311(self): test_retry_wrapper(self._test_get_shadow_failure_invalid_subscription_topic311) def _test_get_shadow_failure_subscription_topic_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_subscription_topic_filter(options))) def test_get_shadow_failure_subscription_topic_type_mismatch5(self): test_retry_wrapper(self._test_get_shadow_failure_subscription_topic_type_mismatch5) def _test_get_shadow_failure_subscription_topic_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_subscription_topic_filter(options))) def test_get_shadow_failure_subscription_topic_type_mismatch311(self): test_retry_wrapper(self._test_get_shadow_failure_subscription_topic_type_mismatch311) def _test_get_shadow_failure_subscriptions_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_subscriptions(options))) def test_get_shadow_failure_subscriptions_type_mismatch5(self): test_retry_wrapper(self._test_get_shadow_failure_subscriptions_type_mismatch5) def _test_get_shadow_failure_subscriptions_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_subscriptions(options))) def test_get_shadow_failure_subscriptions_type_mismatch311(self): test_retry_wrapper(self._test_get_shadow_failure_subscriptions_type_mismatch311) def _test_get_shadow_failure_empty_subscriptions5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _empty_subscription_topic_filters(options))) def test_get_shadow_failure_empty_subscriptions5(self): test_retry_wrapper(self._test_get_shadow_failure_empty_subscriptions5) def _test_get_shadow_failure_empty_subscriptions311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _empty_subscription_topic_filters(options))) def test_get_shadow_failure_empty_subscriptions311(self): test_retry_wrapper(self._test_get_shadow_failure_empty_subscriptions311) def _test_get_shadow_failure_none_publish_topic5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _none_publish_topic(options))) def test_get_shadow_failure_none_publish_topic5(self): test_retry_wrapper(self._test_get_shadow_failure_none_publish_topic5) def _test_get_shadow_failure_none_publish_topic311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _none_publish_topic(options))) def test_get_shadow_failure_none_publish_topic311(self): test_retry_wrapper(self._test_get_shadow_failure_none_publish_topic311) def _test_get_shadow_failure_bad_publish_topic5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_future_failure_test( protocol_client, lambda options: _bad_publish_topic(options))) def test_get_shadow_failure_bad_publish_topic5(self): test_retry_wrapper(self._test_get_shadow_failure_bad_publish_topic5) def _test_get_shadow_failure_bad_publish_topic311(self): self._do_mqtt311_test( lambda protocol_client: self._do_get_shadow_future_failure_test( protocol_client, lambda options: _bad_publish_topic(options))) def test_get_shadow_failure_bad_publish_topic311(self): test_retry_wrapper(self._test_get_shadow_failure_bad_publish_topic311) def _test_get_shadow_failure_publish_topic_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_publish_topic(options))) def test_get_shadow_failure_publish_topic_type_mismatch5(self): test_retry_wrapper(self._test_get_shadow_failure_publish_topic_type_mismatch5) def _test_get_shadow_failure_publish_topic_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_publish_topic(options))) def test_get_shadow_failure_publish_topic_type_mismatch311(self): test_retry_wrapper(self._test_get_shadow_failure_publish_topic_type_mismatch311) def _test_get_shadow_failure_correlation_token_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_correlation_token(options))) def test_get_shadow_failure_correlation_token_type_mismatch5(self): test_retry_wrapper(self._test_get_shadow_failure_correlation_token_type_mismatch5) def _test_get_shadow_failure_correlation_token_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_get_shadow_failure_test( protocol_client, lambda options: _type_mismatch_correlation_token(options))) def test_get_shadow_failure_correlation_token_type_mismatch311(self): test_retry_wrapper(self._test_get_shadow_failure_correlation_token_type_mismatch311) # ============================================================== # streaming operation SUCCESS TEST CASES # ============================================================== def _test_streaming_operation_success5(self): self._do_mqtt5_test(lambda protocol_client: self._do_stream_success_test(protocol_client)) def test_streaming_operation_success5(self): test_retry_wrapper(self._test_streaming_operation_success5) def _test_streaming_operation_success311(self): self._do_mqtt311_test(lambda protocol_client: self._do_stream_success_test(protocol_client)) def test_streaming_operation_success311(self): test_retry_wrapper(self._test_streaming_operation_success311) # ============================================================== # create_stream FAILURE TEST CASES # ============================================================== def _test_create_stream_failure_subscription_topic_filter_none5(self): self._do_mqtt5_test(lambda protocol_client: self._do_create_stream_failure_test( protocol_client, lambda options: _subscription_topic_filter_none(options))) def test_create_stream_failure_subscription_topic_filter_none5(self): test_retry_wrapper(self._test_create_stream_failure_subscription_topic_filter_none5) def _test_create_stream_failure_subscription_topic_filter_none311(self): self._do_mqtt311_test(lambda protocol_client: self._do_create_stream_failure_test( protocol_client, lambda options: _subscription_topic_filter_none(options))) def test_create_stream_failure_subscription_topic_filter_none311(self): test_retry_wrapper(self._test_create_stream_failure_subscription_topic_filter_none311) def _test_create_stream_failure_subscription_topic_filter_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_create_stream_failure_test( protocol_client, lambda options: _type_mismatch_stream_subscription_topic_filter(options))) def test_create_stream_failure_subscription_topic_filter_type_mismatch5(self): test_retry_wrapper(self._test_create_stream_failure_subscription_topic_filter_type_mismatch5) def _test_create_stream_failure_subscription_topic_filter_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_create_stream_failure_test( protocol_client, lambda options: _type_mismatch_stream_subscription_topic_filter(options))) def test_create_stream_failure_subscription_topic_filter_type_mismatch311(self): test_retry_wrapper(self._test_create_stream_failure_subscription_topic_filter_type_mismatch311) def _test_create_stream_failure_subscription_status_listener_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_create_stream_failure_test( protocol_client, lambda options: _type_mismatch_subscription_status_listener(options))) def test_create_stream_failure_subscription_status_listener_type_mismatch5(self): test_retry_wrapper(self._test_create_stream_failure_subscription_status_listener_type_mismatch5) def _test_create_stream_failure_subscription_status_listener_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_create_stream_failure_test( protocol_client, lambda options: _type_mismatch_subscription_status_listener(options))) def test_create_stream_failure_subscription_status_listener_type_mismatch311(self): test_retry_wrapper(self._test_create_stream_failure_subscription_status_listener_type_mismatch311) def _test_create_stream_failure_incoming_publish_listener_type_mismatch5(self): self._do_mqtt5_test(lambda protocol_client: self._do_create_stream_failure_test( protocol_client, lambda options: _type_mismatch_incoming_publish_listener(options))) def test_create_stream_failure_incoming_publish_listener_type_mismatch5(self): test_retry_wrapper(self._test_create_stream_failure_incoming_publish_listener_type_mismatch5) def _test_create_stream_failure_incoming_publish_listener_type_mismatch311(self): self._do_mqtt311_test(lambda protocol_client: self._do_create_stream_failure_test( protocol_client, lambda options: _type_mismatch_incoming_publish_listener(options))) def test_create_stream_failure_incoming_publish_listener_type_mismatch311(self): test_retry_wrapper(self._test_create_stream_failure_incoming_publish_listener_type_mismatch311) if __name__ == 'main': unittest.main()