feature(azure-iot-device): Make python device code compatible internally for digital twin changes in the future

azure-iot-device/azure/iot/device/constant.py

@@ -13,3 +13,8 @@
 PROVISIONING_API_VERSION = "2019-03-31"
 SECURITY_MESSAGE_INTERFACE_ID = "urn:azureiot:Security:SecurityAgent:1"
 TELEMETRY_MESSAGE_SIZE_LIMIT = 262144
+# Everything in digital twin is defined here
+# as things are extremely dynamic and subject to sudden changes
+DIGITAL_TWIN_PREFIX = "dtmi"
+DIGITAL_TWIN_API_VERSION = "2020-05-31-preview"
+DIGITAL_TWIN_QUERY_HEADER = "digital-twin-model-id"

azure-iot-device/azure/iot/device/iothub/pipeline/pipeline_stages_iothub_mqtt.py

@@ -47,15 +47,24 @@ def _run_op(self, op):
                 # Device Format
                 client_id = self.pipeline_root.pipeline_configuration.device_id
 
+            query_param_seq = []
+
             # Apply query parameters (i.e. key1=value1&key2=value2...&keyN=valueN format)
-            query_param_seq = [
-                ("api-version", pkg_constant.IOTHUB_API_VERSION),
-                (
-                    "DeviceClientType",
-                    user_agent.get_iothub_user_agent()
-                    + self.pipeline_root.pipeline_configuration.product_info,
-                ),
-            ]
+            custom_product_info = str(self.pipeline_root.pipeline_configuration.product_info)
+            if custom_product_info.startswith(
+                    pkg_constant.DIGITAL_TWIN_PREFIX
+            ):  # Digital Twin Stuff
+                query_param_seq.append(("api-version", pkg_constant.DIGITAL_TWIN_API_VERSION))
+                query_param_seq.append(("DeviceClientType", user_agent.get_iothub_user_agent()))
+                query_param_seq.append(
+                    (pkg_constant.DIGITAL_TWIN_QUERY_HEADER, custom_product_info)
+                )
+            else:
+                query_param_seq.append(("api-version", pkg_constant.IOTHUB_API_VERSION))
+                query_param_seq.append(
+                    ("DeviceClientType", user_agent.get_iothub_user_agent() + custom_product_info)
+                )
+
             username = "{hostname}/{client_id}/?{query_params}".format(
                 hostname=self.pipeline_root.pipeline_configuration.hostname,
                 client_id=client_id,

azure-iot-device/tests/iothub/pipeline/test_pipeline_stages_iothub_mqtt.py

@@ -173,6 +173,37 @@ def test_username(self, stage, op, pipeline_config, cust_product_info):
         )
         assert op.username == expected_username
 
+    @pytest.mark.it(
+        "Derives the MQTT username, and sets it on the op for digital twin specific scenarios"
+    )
+    @pytest.mark.parametrize(
+        "digital_twin_product_info",
+        [
+            pytest.param(
+                pkg_constant.DIGITAL_TWIN_PREFIX + ":com:example:ClimateSensor;1",
+                id="With custom product info",
+            ),
+            pytest.param(
+                pkg_constant.DIGITAL_TWIN_PREFIX + ":com:example:$Climate$Sensor;1",
+                id="With custom product info (URL encoding required)",
+            ),
+        ],
+    )
+    def test_username_for_digital_twin(self, stage, op, pipeline_config, digital_twin_product_info):
+        pipeline_config.product_info = digital_twin_product_info
+        assert not hasattr(op, "username")
+        stage.run_op(op)
+
+        expected_username = "{hostname}/{client_id}/?api-version={api_version}&DeviceClientType={user_agent}&{digital_twin_prefix}={custom_product_info}".format(
+            hostname=pipeline_config.hostname,
+            client_id=pipeline_config.device_id,
+            api_version=pkg_constant.DIGITAL_TWIN_API_VERSION,
+            user_agent=urllib.parse.quote(user_agent.get_iothub_user_agent(), safe=""),
+            digital_twin_prefix=pkg_constant.DIGITAL_TWIN_QUERY_HEADER,
+            custom_product_info=urllib.parse.quote(pipeline_config.product_info, safe=""),
+        )
+        assert op.username == expected_username
+
     @pytest.mark.it(
         "ALWAYS uses the pipeline configuration's hostname in the MQTT username and NEVER the gateway_hostname"
     )
@@ -249,6 +280,40 @@ def test_username(self, stage, op, pipeline_config, cust_product_info):
         )
         assert op.username == expected_username
 
+    @pytest.mark.it(
+        "Derives the MQTT username, and sets it on the op for digital twin specific scenarios"
+    )
+    @pytest.mark.parametrize(
+        "digital_twin_product_info",
+        [
+            pytest.param(
+                pkg_constant.DIGITAL_TWIN_PREFIX + ":com:example:ClimateSensor;1",
+                id="With custom product info",
+            ),
+            pytest.param(
+                pkg_constant.DIGITAL_TWIN_PREFIX + ":com:example:$Climate$Sensor;1",
+                id="With custom product info (URL encoding required)",
+            ),
+        ],
+    )
+    def test_username_for_digital_twin(self, stage, op, pipeline_config, digital_twin_product_info):
+        pipeline_config.product_info = digital_twin_product_info
+        assert not hasattr(op, "username")
+        stage.run_op(op)
+
+        expected_client_id = "{device_id}/{module_id}".format(
+            device_id=pipeline_config.device_id, module_id=pipeline_config.module_id
+        )
+        expected_username = "{hostname}/{client_id}/?api-version={api_version}&DeviceClientType={user_agent}&{digital_twin_prefix}={custom_product_info}".format(
+            hostname=pipeline_config.hostname,
+            client_id=expected_client_id,
+            api_version=pkg_constant.DIGITAL_TWIN_API_VERSION,
+            user_agent=urllib.parse.quote(user_agent.get_iothub_user_agent(), safe=""),
+            digital_twin_prefix=pkg_constant.DIGITAL_TWIN_QUERY_HEADER,
+            custom_product_info=urllib.parse.quote(pipeline_config.product_info, safe=""),
+        )
+        assert op.username == expected_username
+
     @pytest.mark.it(
         "ALWAYS uses the pipeline configuration's hostname in the MQTT username and NEVER the gateway_hostname"
     )