DT Help + test improvement.

azext_iot/digitaltwins/_help.py

@@ -19,7 +19,7 @@ def load_digitaltwins_help():
 
     helps["dt create"] = """
         type: command
-        short-summary: Create a new Digital Twins instance.
+        short-summary: Create or update a Digital Twins instance.
 
         examples:
         - name: Create instance in target resource group using the resource group location.
@@ -50,6 +50,18 @@ def load_digitaltwins_help():
             "/subscriptions/a12345ea-bb21-994d-2263-c716348e32a1/resourceGroups/ProResourceGroup/providers/Microsoft.EventHub/namespaces/myEventHubNamespace/eventhubs/myEventHub"
             "/subscriptions/a12345ea-bb21-994d-2263-c716348e32a1/resourceGroups/ProResourceGroup/providers/Microsoft.ServiceBus/namespaces/myServiceBusNamespace/topics/myTopic"
             --role MyCustomRole
+
+        - name: Update an instance in the target resource group to enable system managed identity.
+          text: >
+            az dt create -n {instance_name} -g {resouce_group} --assign-identity
+
+        - name: Update an instance in the target resource group to disable system managed identity.
+          text: >
+            az dt create -n {instance_name} -g {resouce_group} --assign-identity false
+
+        - name: Update an instance in the target resource group with new tag values and disable public network access.
+          text: >
+            az dt create -n {instance_name} -g {resouce_group} --tags env=prod --public-network-access Disabled
     """
 
     helps["dt show"] = """

azext_iot/tests/digitaltwins/__init__.py

@@ -160,7 +160,7 @@ def tearDown(self):
 
     # Needed because the DT service will indicate provisioning is finished before it actually is.
     def wait_for_hostname(
-        self, instance: dict, wait_in_sec: int = 5, interval: int = 3
+        self, instance: dict, wait_in_sec: int = 5, interval: int = 4
     ):
         from time import sleep
 
@@ -176,7 +176,7 @@ def wait_for_hostname(
                 )
             ).as_json()
 
-            if refereshed_instance.get("hostName"):
+            if refereshed_instance.get("hostName") and refereshed_instance["provisioningState"] == "Succeeded":
                 return refereshed_instance
 
         return instance

azext_iot/tests/digitaltwins/test_dt_privatelinks_lifecycle_int.py

@@ -28,11 +28,22 @@ def test_dt_privatelinks(self):
             )
         ).get_output_in_json()
         self.track_instance(create_output)
+        create_output = self.wait_for_hostname(create_output)
 
         # Fail test if hostName missing
         assert create_output.get(
             "hostName"
         ), "Service failed to provision DT instance: {}.".format(instance_name)
+        assert create_output["publicNetworkAccess"] == "Enabled"
+
+        update_output = self.cmd(
+            "dt create -n {} -g {} -l {} --public-network-access Disabled".format(
+                instance_name,
+                self.rg,
+                self.region,
+            )
+        ).get_output_in_json()
+        assert update_output["publicNetworkAccess"] == "Disabled"
 
         list_priv_links = self.cmd(
             "dt network private-link list -n {} -g {}".format(

azext_iot/tests/digitaltwins/test_dt_resource_lifecycle_int.py

@@ -130,6 +130,22 @@ def test_dt_resource(self):
             assign_identity=True,
         )
 
+        # Update tags and disable MSI
+        updated_tags = "env=test tier=premium"
+        updated_tags_dict = {"env": "test", "tier": "premium"}
+        remove_msi_output = self.cmd(
+            "dt create -n {} -g {} --assign-identity false --tags {}".format(instance_names[1], self.rg, updated_tags)
+        ).get_output_in_json()
+
+        assert_common_resource_attributes(
+            self.wait_for_hostname(remove_msi_output),
+            instance_names[1],
+            self.rg,
+            self.rg_region,
+            tags=updated_tags_dict,
+            assign_identity=False,
+        )
+
         list_output = self.cmd("dt list").get_output_in_json()
         filtered_list = filter_dt_list(list_output, instance_names)
         assert len(filtered_list) == len(instance_names)