Device Advisor Test Samples

deviceadvisor/tests/da_test_utils.py

@@ -0,0 +1,33 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+import os
+from enum import Enum
+from uuid import uuid4
+
+class TestType(Enum):
+    CONNECT = 1
+    SUB_PUB = 1
+    SHADOW = 1
+
+class DATestUtils:
+    endpoint = os.getenv('DA_ENDPOINT')
+    certificatePath = os.getenv('DA_CERTI')
+    keyPath = os.getenv('DA_KEY')
+    topic = os.getenv('DA_TOPIC')
+    thing_name = os.getenv('DA_THING_NAME')
+    shadowProperty = os.getenv('DA_SHADOW_PROPERTY')
+    shadowValue = os.getenv('DA_SHADOW_VALUE_SET')
+    client_id = "test-" + str(uuid4())
+
+    @classmethod
+    def valid(cls, test_type):
+        if (not (cls.endpoint and cls.certificatePath and cls.keyPath)):
+            return False
+
+        if (not cls.topic and test_type == TestType.SUB_PUB):
+            return False
+
+        if (not (cls.thing_name and cls.shadowProperty and cls.shadowValue) and test_type == TestType.SHADOW):
+            return False
+
+        return True

deviceadvisor/tests/mqtt_connect.py

@@ -0,0 +1,28 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+
+from awsiot import mqtt_connection_builder
+from da_test_utils import DATestUtils, TestType
+
+# This sample used to test mqtt connection through device advisor
+
+if __name__ == '__main__':
+    # init variables
+    utils = DATestUtils.valid(TestType.CONNECT)
+    if not utils:
+        quit(-1)
+
+    mqtt_connection = mqtt_connection_builder.mtls_from_path(
+        endpoint=DATestUtils.endpoint,
+        cert_filepath=DATestUtils.certificatePath,
+        pri_key_filepath=DATestUtils.keyPath,
+        client_id = DATestUtils.client_id)
+
+    connect_future = mqtt_connection.connect()
+
+    # Future.result() waits until a result is available
+    connect_future.result()
+
+    disconnect_future = mqtt_connection.disconnect()
+    disconnect_future.result()
+    quit(0)

deviceadvisor/tests/mqtt_publish.py

@@ -0,0 +1,37 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+from awscrt import mqtt
+from awsiot import mqtt_connection_builder
+from da_test_utils import DATestUtils, TestType
+import json
+
+if __name__ == '__main__':
+    # validate environment variables
+    utils = DATestUtils.valid(TestType.SUB_PUB)
+    if not utils:
+        quit(-1)
+
+    mqtt_connection = mqtt_connection_builder.mtls_from_path(
+        endpoint=DATestUtils.endpoint,
+        cert_filepath=DATestUtils.certificatePath,
+        pri_key_filepath=DATestUtils.keyPath,
+        client_id=DATestUtils.client_id)
+    connect_future = mqtt_connection.connect()
+
+    # Future.result() waits until a result is available
+    connect_future.result()
+
+    message = "Hello World"
+    message_json = json.dumps(message)
+    # Device advisor test will not return PUBACK, therefore we use AT_MOST_ONCE so that 
+    # we dont busy wait for PUBACK
+    publish_future, packet_id = mqtt_connection.publish(
+        topic=DATestUtils.topic,
+        payload=message_json,
+        qos=mqtt.QoS.AT_MOST_ONCE)
+    publish_future.result()
+
+    # Disconnect
+    disconnect_future = mqtt_connection.disconnect()
+    disconnect_future.result()
+    quit(0)

deviceadvisor/tests/mqtt_subscribe.py

@@ -0,0 +1,35 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+
+from awscrt import io, mqtt, auth, http
+from awsiot import mqtt_connection_builder
+from da_test_utils import DATestUtils, TestType
+
+if __name__ == '__main__':
+    # init variables
+    utils = DATestUtils.valid(TestType.SUB_PUB)
+    if not utils:
+        quit(-1)
+
+    mqtt_connection = mqtt_connection_builder.mtls_from_path(
+        endpoint=DATestUtils.endpoint,
+        cert_filepath=DATestUtils.certificatePath,
+        pri_key_filepath=DATestUtils.keyPath,
+        client_id = DATestUtils.client_id)
+
+    connect_future = mqtt_connection.connect()
+
+    # Future.result() waits until a result is available
+    connect_future.result()
+
+    # Subscribe
+    subscribe_future, packet_id = mqtt_connection.subscribe(
+        topic=DATestUtils.topic,
+        qos=mqtt.QoS.AT_LEAST_ONCE)
+
+    subscribe_future.result()
+
+    # Disconnect
+    disconnect_future = mqtt_connection.disconnect()
+    disconnect_future.result()
+    quit(0)

deviceadvisor/tests/shadow_update.py

@@ -0,0 +1,42 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+
+from awscrt import mqtt
+from awsiot import iotshadow, mqtt_connection_builder
+from concurrent.futures import Future
+from da_test_utils import DATestUtils, TestType
+from uuid import uuid4
+
+if __name__ == '__main__':
+    # init variables
+    utils = DATestUtils.valid(TestType.SUB_PUB)
+    if not utils:
+        quit(-1)
+
+    mqtt_connection = mqtt_connection_builder.mtls_from_path(
+        endpoint=DATestUtils.endpoint,
+        cert_filepath=DATestUtils.certificatePath,
+        pri_key_filepath=DATestUtils.keyPath,
+        client_id = DATestUtils.client_id)
+
+    connect_future = mqtt_connection.connect()
+    connect_future.result()
+    shadow_client = iotshadow.IotShadowClient(mqtt_connection)
+
+
+    # Publish shadow value
+    request = iotshadow.UpdateShadowRequest(
+        thing_name=DATestUtils.thing_name,
+        state=iotshadow.ShadowState(
+            reported={ DATestUtils.shadowProperty: DATestUtils.shadowValue },
+            desired={ DATestUtils.shadowProperty: DATestUtils.shadowValue },
+        )
+    )
+    # Device advisor test will not return PUBACK, therefore we use AT_MOST_ONCE so that 
+    # we dont busy wait for PUBACK
+    shadow_future = shadow_client.publish_update_shadow(request, mqtt.QoS.AT_MOST_ONCE)
+    shadow_future.result()
+
+    disconnect_future = mqtt_connection.disconnect()
+    disconnect_future.result()
+    quit(0)