diff --git a/azext_iot/digitaltwins/_help.py b/azext_iot/digitaltwins/_help.py
index 2ebb5f69e..10dd5c1ae 100644
--- a/azext_iot/digitaltwins/_help.py
+++ b/azext_iot/digitaltwins/_help.py
@@ -209,6 +209,95 @@ def load_digitaltwins_help():
             az dt endpoint delete -n {instance_name} --endpoint-name {endpoint_name} -y --no-wait
     """
 
+    helps["dt network"] = """
+        type: group
+        short-summary: Manage Digital Twins network configuration including private links and endpoint connections.
+    """
+
+    helps["dt network private-link"] = """
+        type: group
+        short-summary: Manage Digital Twins instance private-link operations.
+    """
+
+    helps["dt network private-link show"] = """
+        type: command
+        short-summary: Show a private-link associated with the instance.
+
+        examples:
+        - name: Show the private-link named "API" associated with the instance.
+          text: >
+            az dt network private-link show -n {instance_name} --link-name API
+    """
+
+    helps["dt network private-link list"] = """
+        type: command
+        short-summary: List private-links associated with the Digital Twins instance.
+
+        examples:
+        - name: List all private-links associated with the instance.
+          text: >
+            az dt network private-link list -n {instance_name}
+    """
+
+    helps["dt network private-endpoint"] = """
+        type: group
+        short-summary: Manage Digital Twins instance private-endpoints.
+        long-summary: Use 'az network private-endpoint create' to create a private-endpoint and link to a Digital Twins resource.
+    """
+
+    helps["dt network private-endpoint connection"] = """
+        type: group
+        short-summary: Manage Digital Twins instance private-endpoint connections.
+    """
+
+    helps["dt network private-endpoint connection list"] = """
+        type: command
+        short-summary: List private-endpoint connections associated with the Digital Twins instance.
+
+        examples:
+        - name: List all private-endpoint connections associated with the instance.
+          text: >
+            az dt network private-endpoint connection list -n {instance_name}
+    """
+
+    helps["dt network private-endpoint connection show"] = """
+        type: command
+        short-summary: Show a private-endpoint connection associated with the Digital Twins instance.
+
+        examples:
+        - name: Show details of the private-endpoint connection named ba8408b6-1372-41b2-aef8-af43afc4729f.
+          text: >
+            az dt network private-endpoint connection show -n {instance_name} --cn ba8408b6-1372-41b2-aef8-af43afc4729f
+    """
+
+    helps["dt network private-endpoint connection set"] = """
+        type: command
+        short-summary: Set the state of a private-endpoint connection associated with the Digital Twins instance.
+
+        examples:
+        - name: Approve a pending private-endpoint connection associated with the instance and add a description.
+          text: >
+            az dt network private-endpoint connection set -n {instance_name} --cn {connection_name} --status Approved --desc "A description."
+
+        - name: Reject a private-endpoint connection associated with the instance and add a description.
+          text: >
+            az dt network private-endpoint connection set -n {instance_name} --cn {connection_name} --status Rejected --desc "Does not comply."
+    """
+
+    helps["dt network private-endpoint connection delete"] = """
+        type: command
+        short-summary: Delete a private-endpoint connection associated with the Digital Twins instance.
+
+        examples:
+        - name: Delete the private-endpoint connection named ba8408b6-1372-41b2-aef8-af43afc4729f with confirmation. Block until finished.
+          text: >
+            az dt network private-endpoint connection delete -n {instance_name} --cn ba8408b6-1372-41b2-aef8-af43afc4729f
+
+        - name: Delete the private-endpoint connection named ba8408b6-1372-41b2-aef8-af43afc4729f no confirmation. Return immediately.
+          text: >
+            az dt network private-endpoint connection delete -n {instance_name} --cn ba8408b6-1372-41b2-aef8-af43afc4729f -y --no-wait
+    """
+
     helps["dt role-assignment"] = """
         type: group
         short-summary: Manage RBAC role assignments for a Digital Twins instance.
diff --git a/azext_iot/digitaltwins/command_map.py b/azext_iot/digitaltwins/command_map.py
index 803e00a8c..3b7df0b27 100644
--- a/azext_iot/digitaltwins/command_map.py
+++ b/azext_iot/digitaltwins/command_map.py
@@ -4,8 +4,6 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-from azure.cli.core.profiles import ResourceType
-
 """
 Load CLI commands
 """
@@ -39,7 +37,6 @@ def load_digitaltwins_commands(self, _):
     with self.command_group(
         "dt",
         command_type=digitaltwins_resource_ops,
-        resource_type=ResourceType.MGMT_RESOURCE_RESOURCES,
     ) as cmd_group:
         cmd_group.command("create", "create_instance")
         cmd_group.show_command("show", "show_instance")
@@ -140,3 +137,31 @@ def load_digitaltwins_commands(self, _):
         )
         cmd_group.command("update", "update_model")
         cmd_group.command("delete", "delete_model")
+
+    with self.command_group(
+        "dt network",
+        command_type=digitaltwins_resource_ops,
+    ) as cmd_group:
+        pass
+
+    with self.command_group(
+        "dt network private-link",
+        command_type=digitaltwins_resource_ops,
+    ) as cmd_group:
+        cmd_group.show_command("show", "show_private_link")
+        cmd_group.command("list", "list_private_links")
+
+    with self.command_group(
+        "dt network private-endpoint",
+        command_type=digitaltwins_resource_ops,
+    ) as cmd_group:
+        pass
+
+    with self.command_group(
+        "dt network private-endpoint connection",
+        command_type=digitaltwins_resource_ops,
+    ) as cmd_group:
+        cmd_group.command("set", "set_private_endpoint_conn")
+        cmd_group.show_command("show", "show_private_endpoint_conn")
+        cmd_group.command("list", "list_private_endpoint_conns")
+        cmd_group.command("delete", "delete_private_endpoint_conn", confirmation=True, supports_no_wait=True)
diff --git a/azext_iot/digitaltwins/commands_resource.py b/azext_iot/digitaltwins/commands_resource.py
index fc889415d..96d176bc0 100644
--- a/azext_iot/digitaltwins/commands_resource.py
+++ b/azext_iot/digitaltwins/commands_resource.py
@@ -5,7 +5,11 @@
 # --------------------------------------------------------------------------------------------
 
 from azext_iot.digitaltwins.providers.resource import ResourceProvider
-from azext_iot.digitaltwins.common import ADTEndpointType, ADTEndpointAuthType
+from azext_iot.digitaltwins.common import (
+    ADTEndpointType,
+    ADTEndpointAuthType,
+    ADTPublicNetworkAccessType,
+)
 from knack.log import get_logger
 
 logger = get_logger(__name__)
@@ -20,6 +24,7 @@ def create_instance(
     assign_identity=None,
     scopes=None,
     role_type="Contributor",
+    public_network_access=ADTPublicNetworkAccessType.enabled.value,
 ):
     rp = ResourceProvider(cmd)
     return rp.create(
@@ -30,6 +35,7 @@ def create_instance(
         assign_identity=assign_identity,
         scopes=scopes,
         role_type=role_type,
+        public_network_access=public_network_access,
     )
 
 
@@ -157,3 +163,58 @@ def add_endpoint_eventhub(
         dead_letter_secret=dead_letter_secret,
         auth_type=auth_type,
     )
+
+
+def show_private_link(cmd, name, link_name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.get_private_link(
+        name=name, resource_group_name=resource_group_name, link_name=link_name
+    )
+
+
+def list_private_links(cmd, name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.list_private_links(name=name, resource_group_name=resource_group_name)
+
+
+def set_private_endpoint_conn(
+    cmd,
+    name,
+    conn_name,
+    status,
+    description=None,
+    group_ids=None,
+    actions_required=None,
+    resource_group_name=None,
+):
+    rp = ResourceProvider(cmd)
+    return rp.set_private_endpoint_conn(
+        name=name,
+        resource_group_name=resource_group_name,
+        conn_name=conn_name,
+        status=status,
+        description=description,
+        group_ids=group_ids,
+        actions_required=actions_required,
+    )
+
+
+def show_private_endpoint_conn(cmd, name, conn_name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.get_private_endpoint_conn(
+        name=name, resource_group_name=resource_group_name, conn_name=conn_name
+    )
+
+
+def list_private_endpoint_conns(cmd, name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.list_private_endpoint_conns(
+        name=name, resource_group_name=resource_group_name
+    )
+
+
+def delete_private_endpoint_conn(cmd, name, conn_name, resource_group_name=None):
+    rp = ResourceProvider(cmd)
+    return rp.delete_private_endpoint_conn(
+        name=name, resource_group_name=resource_group_name, conn_name=conn_name
+    )
diff --git a/azext_iot/digitaltwins/common.py b/azext_iot/digitaltwins/common.py
index b0e5cb0a3..b0a382a70 100644
--- a/azext_iot/digitaltwins/common.py
+++ b/azext_iot/digitaltwins/common.py
@@ -14,7 +14,7 @@
 
 class ADTEndpointType(Enum):
     """
-    ADT Endpoint Type.
+    ADT endpoint type.
     """
 
     eventgridtopic = "eventgridtopic"
@@ -24,8 +24,28 @@ class ADTEndpointType(Enum):
 
 class ADTEndpointAuthType(Enum):
     """
-    ADT Endpoint Auth Type.
+    ADT endpoint auth type.
     """
 
     identitybased = "IdentityBased"
     keybased = "KeyBased"
+
+
+class ADTPrivateConnectionStatusType(Enum):
+    """
+    ADT private endpoint connection status type.
+    """
+
+    pending = "Pending"
+    approved = "Approved"
+    rejected = "Rejected"
+    disconnected = "Disconnected"
+
+
+class ADTPublicNetworkAccessType(Enum):
+    """
+    ADT private endpoint connection status type.
+    """
+
+    enabled = "Enabled"
+    disabled = "Disabled"
diff --git a/azext_iot/digitaltwins/params.py b/azext_iot/digitaltwins/params.py
index 603d8a0a5..f9a217107 100644
--- a/azext_iot/digitaltwins/params.py
+++ b/azext_iot/digitaltwins/params.py
@@ -15,7 +15,11 @@
     get_enum_type,
     tags_type,
 )
-from azext_iot.digitaltwins.common import ADTEndpointAuthType
+from azext_iot.digitaltwins.common import (
+    ADTEndpointAuthType,
+    ADTPrivateConnectionStatusType,
+    ADTPublicNetworkAccessType,
+)
 
 depfor_type = CLIArgumentType(
     options_list=["--dependencies-for"],
@@ -72,10 +76,14 @@ def load_digitaltwins_arguments(self, _):
             help="Event route name.",
         )
         context.argument(
-            "filter", options_list=["--filter"], help="Event route filter.",
+            "filter",
+            options_list=["--filter"],
+            help="Event route filter.",
         )
         context.argument(
-            "role_type", options_list=["--role"], help="Role name or Id.",
+            "role_type",
+            options_list=["--role"],
+            help="Role name or Id.",
         )
         context.argument(
             "assignee",
@@ -89,7 +97,9 @@ def load_digitaltwins_arguments(self, _):
             help="Digital Twins model Id. Example: dtmi:com:example:Room;2",
         )
         context.argument(
-            "twin_id", options_list=["--twin-id", "-t"], help="The digital twin Id.",
+            "twin_id",
+            options_list=["--twin-id", "-t"],
+            help="The digital twin Id.",
         )
         context.argument(
             "include_inherited",
@@ -103,6 +113,13 @@ def load_digitaltwins_arguments(self, _):
             options_list=["--top"],
             help="Maximum number of elements to return.",
         )
+        context.argument(
+            "public_network_access",
+            options_list=["--public-network-access", "--pna"],
+            help="Determines if the Digital Twins instance can be accessed from a public network.",
+            arg_group="Networking",
+            arg_type=get_enum_type(ADTPublicNetworkAccessType),
+        )
 
     with self.argument_context("dt create") as context:
         context.argument(
@@ -130,19 +147,19 @@ def load_digitaltwins_arguments(self, _):
             "dead_letter_secret",
             options_list=["--deadletter-sas-uri", "--dsu"],
             help="Dead-letter storage container URL with SAS token for Key based authentication.",
-            arg_group="Dead-letter Endpoint"
+            arg_group="Dead-letter Endpoint",
         )
         context.argument(
             "dead_letter_uri",
             options_list=["--deadletter-uri", "--du"],
             help="Dead-letter storage container URL for Identity based authentication.",
-            arg_group="Dead-letter Endpoint"
+            arg_group="Dead-letter Endpoint",
         )
         context.argument(
             "auth_type",
             options_list=["--auth-type"],
             help="Endpoint authentication type.",
-            arg_type=get_enum_type(ADTEndpointAuthType)
+            arg_type=get_enum_type(ADTEndpointAuthType),
         )
 
     with self.argument_context("dt endpoint create eventgrid") as context:
@@ -352,5 +369,48 @@ def load_digitaltwins_arguments(self, _):
             help="Indicates intent to decommission a target model.",
         )
         context.argument(
-            "dependencies_for", arg_type=depfor_type,
+            "dependencies_for",
+            arg_type=depfor_type,
+        )
+
+    with self.argument_context("dt network private-link") as context:
+        context.argument(
+            "link_name",
+            options_list=["--link-name", "--ln"],
+            help="Private link name.",
+            arg_group="Private Connection",
+        )
+
+    with self.argument_context("dt network private-endpoint") as context:
+        context.argument(
+            "conn_name",
+            options_list=["--conn-name", "--cn"],
+            help="Private endpoint connection name.",
+            arg_group="Private-Endpoint",
+        )
+        context.argument(
+            "group_ids",
+            options_list=["--group-ids"],
+            help="Space seperated list of group ids that the private endpoint should connect to.",
+            arg_group="Private-Endpoint",
+            nargs="+",
+        )
+        context.argument(
+            "status",
+            options_list=["--status"],
+            help="The status of a private endpoint connection.",
+            arg_type=get_enum_type(ADTPrivateConnectionStatusType),
+            arg_group="Private-Endpoint",
+        )
+        context.argument(
+            "description",
+            options_list=["--description", "--desc"],
+            help="Description for the private endpoint connection.",
+            arg_group="Private-Endpoint",
+        )
+        context.argument(
+            "actions_required",
+            options_list=["--actions-required", "--ar"],
+            help="A message indicating if changes on the service provider require any updates on the consumer.",
+            arg_group="Private-Endpoint",
         )
diff --git a/azext_iot/digitaltwins/providers/resource.py b/azext_iot/digitaltwins/providers/resource.py
index c2cb29259..f643764e7 100644
--- a/azext_iot/digitaltwins/providers/resource.py
+++ b/azext_iot/digitaltwins/providers/resource.py
@@ -3,14 +3,19 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
-from azext_iot.digitaltwins.common import ADTEndpointAuthType
+from azext_iot.digitaltwins.common import (
+    ADTEndpointAuthType,
+    ADTPublicNetworkAccessType,
+)
 from azext_iot.digitaltwins.providers import (
     DigitalTwinsResourceManager,
     CloudError,
     ErrorResponseException,
 )
 from azext_iot.digitaltwins.providers.rbac import RbacProvider
-from azext_iot.sdk.digitaltwins.controlplane.models import DigitalTwinsDescription
+from azext_iot.sdk.digitaltwins.controlplane.models import (
+    DigitalTwinsDescription,
+)
 from azext_iot.common.utility import unpack_msrest_error
 from knack.util import CLIError
 from knack.log import get_logger
@@ -34,6 +39,7 @@ def create(
         assign_identity=None,
         scopes=None,
         role_type="Contributor",
+        public_network_access=ADTPublicNetworkAccessType.enabled.value,
     ):
         if not location:
             from azext_iot.common.embedded_cli import EmbeddedCLI
@@ -56,6 +62,7 @@ def create(
                 location=location,
                 tags=tags,
                 identity={"type": "SystemAssigned" if assign_identity else "None"},
+                public_network_access=public_network_access,
             )
             create_or_update = self.mgmt_sdk.digital_twins.create_or_update(
                 resource_name=name,
@@ -342,3 +349,119 @@ def add_endpoint(
             )
         except ErrorResponseException as e:
             raise CLIError(unpack_msrest_error(e))
+
+    def get_private_link(self, name, link_name, resource_group_name=None):
+        target_instance = self.find_instance(
+            name=name, resource_group_name=resource_group_name
+        )
+        if not resource_group_name:
+            resource_group_name = self.get_rg(target_instance)
+
+        try:
+            return self.mgmt_sdk.private_link_resources.get(
+                resource_group_name=resource_group_name,
+                resource_name=name,
+                resource_id=link_name,
+                raw=True,
+            ).response.json()
+        except ErrorResponseException as e:
+            raise CLIError(unpack_msrest_error(e))
+
+    def list_private_links(self, name, resource_group_name=None):
+        target_instance = self.find_instance(
+            name=name, resource_group_name=resource_group_name
+        )
+        if not resource_group_name:
+            resource_group_name = self.get_rg(target_instance)
+
+        try:
+            # This resource is not paged though it may have been the intent.
+            link_collection = self.mgmt_sdk.private_link_resources.list(
+                resource_group_name=resource_group_name, resource_name=name, raw=True
+            ).response.json()
+            return link_collection.get("value", [])
+        except ErrorResponseException as e:
+            raise CLIError(unpack_msrest_error(e))
+
+    def set_private_endpoint_conn(
+        self,
+        name,
+        conn_name,
+        status,
+        description,
+        actions_required=None,
+        group_ids=None,
+        resource_group_name=None,
+    ):
+        target_instance = self.find_instance(
+            name=name, resource_group_name=resource_group_name
+        )
+        if not resource_group_name:
+            resource_group_name = self.get_rg(target_instance)
+
+        try:
+            return self.mgmt_sdk.private_endpoint_connections.create_or_update(
+                resource_group_name=resource_group_name,
+                resource_name=name,
+                private_endpoint_connection_name=conn_name,
+                properties={
+                    "privateLinkServiceConnectionState": {
+                        "status": status,
+                        "description": description,
+                        "actions_required": actions_required,
+                    },
+                    "groupIds": group_ids,
+                },
+            )
+
+        except ErrorResponseException as e:
+            raise CLIError(unpack_msrest_error(e))
+
+    def get_private_endpoint_conn(self, name, conn_name, resource_group_name=None):
+        target_instance = self.find_instance(
+            name=name, resource_group_name=resource_group_name
+        )
+        if not resource_group_name:
+            resource_group_name = self.get_rg(target_instance)
+
+        try:
+            return self.mgmt_sdk.private_endpoint_connections.get(
+                resource_group_name=resource_group_name,
+                resource_name=name,
+                private_endpoint_connection_name=conn_name,
+                raw=True,
+            ).response.json()
+        except ErrorResponseException as e:
+            raise CLIError(unpack_msrest_error(e))
+
+    def list_private_endpoint_conns(self, name, resource_group_name=None):
+        target_instance = self.find_instance(
+            name=name, resource_group_name=resource_group_name
+        )
+        if not resource_group_name:
+            resource_group_name = self.get_rg(target_instance)
+
+        try:
+            # This resource is not paged though it may have been the intent.
+            endpoint_collection = self.mgmt_sdk.private_endpoint_connections.list(
+                resource_group_name=resource_group_name, resource_name=name, raw=True
+            ).response.json()
+            return endpoint_collection.get("value", [])
+        except ErrorResponseException as e:
+            raise CLIError(unpack_msrest_error(e))
+
+    def delete_private_endpoint_conn(self, name, conn_name, resource_group_name=None):
+        target_instance = self.find_instance(
+            name=name, resource_group_name=resource_group_name
+        )
+        if not resource_group_name:
+            resource_group_name = self.get_rg(target_instance)
+
+        try:
+            return self.mgmt_sdk.private_endpoint_connections.delete(
+                resource_group_name=resource_group_name,
+                resource_name=name,
+                private_endpoint_connection_name=conn_name
+            )
+        except ErrorResponseException as e:
+            raise CLIError(unpack_msrest_error(e))
diff --git a/azext_iot/tests/digitaltwins/test_dt_privatelinks_lifecycle_int.py b/azext_iot/tests/digitaltwins/test_dt_privatelinks_lifecycle_int.py
new file mode 100644
index 000000000..c051b3983
--- /dev/null
+++ b/azext_iot/tests/digitaltwins/test_dt_privatelinks_lifecycle_int.py
@@ -0,0 +1,186 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import pytest
+from knack.log import get_logger
+from azext_iot.tests.settings import DynamoSettings
+from . import DTLiveScenarioTest
+from . import generate_resource_id, generate_generic_id
+
+logger = get_logger(__name__)
+
+resource_test_env_vars = ["azext_dt_vnet_subnet_id"]
+
+settings = DynamoSettings(opt_env_set=resource_test_env_vars)
+
+
+class TestDTPrivateLinksLifecycle(DTLiveScenarioTest):
+    def __init__(self, test_case):
+        super(TestDTPrivateLinksLifecycle, self).__init__(test_case)
+
+    @pytest.mark.skipif(
+        not settings.env.azext_dt_vnet_subnet_id,
+        reason="Set azext_dt_vnet_subnet_id (fully qualified resource Id) for private-link/private-endpoint tests.",
+    )
+    def test_dt_privatelinks(self):
+        self.wait_for_capacity()
+
+        instance_name = generate_resource_id()
+        group_id = "API"
+        create_output = self.cmd(
+            "dt create -n {} -g {} -l {}".format(
+                instance_name,
+                self.rg,
+                self.region,
+            )
+        ).get_output_in_json()
+        self.track_instance(create_output)
+
+        # Fail test if hostName missing
+        assert create_output.get(
+            "hostName"
+        ), "Service failed to provision DT instance: {}.".format(instance_name)
+
+        list_priv_links = self.cmd(
+            "dt network private-link list -n {} -g {}".format(
+                instance_name,
+                self.rg,
+            )
+        ).get_output_in_json()
+        assert len(list_priv_links) > 0
+
+        show_api_priv_link = self.cmd(
+            "dt network private-link show -n {} -g {} --ln {}".format(
+                instance_name, self.rg, group_id
+            )
+        ).get_output_in_json()
+        assert show_api_priv_link["name"] == group_id
+        assert (
+            show_api_priv_link["type"]
+            == "Microsoft.DigitalTwins/digitalTwinsInstances/privateLinkResources"
+        )
+
+        connection_name = generate_generic_id()
+        endpoint_name = generate_generic_id()
+        dt_instance_id = create_output["id"]
+        vnet_name = generate_generic_id()
+        subnet_name = generate_generic_id()
+
+        # Create VNET
+        self.cmd(
+            "network vnet create -n {} -g {} --subnet-name {}".format(
+                vnet_name, self.rg, subnet_name
+            ),
+            checks=self.check("length(newVNet.subnets)", 1),
+        )
+        self.cmd(
+            "network vnet subnet update -n {} --vnet-name {} -g {} "
+            "--disable-private-endpoint-network-policies true".format(
+                subnet_name, vnet_name, self.rg
+            ),
+            checks=self.check("privateEndpointNetworkPolicies", "Disabled"),
+        )
+
+        create_priv_endpoint_result = self.embedded_cli.invoke(
+            "network private-endpoint create --connection-name {} -n {} --private-connection-resource-id '{}'"
+            " --group-id {} -g {} --vnet-name {} --subnet {} --manual-request".format(
+                connection_name,
+                endpoint_name,
+                dt_instance_id,
+                group_id,
+                self.rg,
+                vnet_name,
+                subnet_name,
+            )
+        )
+
+        if not create_priv_endpoint_result.success():
+            raise RuntimeError(
+                "Failed to configure private-endpoint for DT instance: {}".format(
+                    instance_name
+                )
+            )
+
+        list_priv_endpoints = self.cmd(
+            "dt network private-endpoint connection list -n {} -g {}".format(
+                instance_name,
+                self.rg,
+            )
+        ).get_output_in_json()
+        assert len(list_priv_endpoints) > 0
+
+        instance_connection_id = list_priv_endpoints[-1]["name"]
+
+        show_priv_endpoint = self.cmd(
+            "dt network private-endpoint connection show -n {} -g {} --cn {}".format(
+                instance_name, self.rg, instance_connection_id
+            )
+        ).get_output_in_json()
+        assert show_priv_endpoint["name"] == instance_connection_id
+        assert (
+            show_priv_endpoint["type"]
+            == "Microsoft.DigitalTwins/digitalTwinsInstances/privateEndpointConnections"
+        )
+        assert show_priv_endpoint["properties"]["provisioningState"] == "Succeeded"
+
+        # Force manual approval
+        assert (
+            show_priv_endpoint["properties"]["privateLinkServiceConnectionState"]["status"]
+            == "Pending"
+        )
+
+        random_desc_approval = "{} {}".format(
+            generate_generic_id(), generate_generic_id()
+        )
+        set_connection_output = self.cmd(
+            "dt network private-endpoint connection set -n {} -g {} --cn {} --status Approved --desc '{}'".format(
+                instance_name, self.rg, instance_connection_id, random_desc_approval
+            )
+        ).get_output_in_json()
+        assert (
+            set_connection_output["properties"]["privateLinkServiceConnectionState"]["status"]
+            == "Approved"
+        )
+        assert (
+            set_connection_output["properties"]["privateLinkServiceConnectionState"]["description"]
+            == random_desc_approval
+        )
+
+        random_desc_rejected = "{} {}".format(
+            generate_generic_id(), generate_generic_id()
+        )
+        set_connection_output = self.cmd(
+            "dt network private-endpoint connection set -n {} -g {} --cn {} --status Rejected --desc '{}'".format(
+                instance_name, self.rg, instance_connection_id, random_desc_rejected
+            )
+        ).get_output_in_json()
+        assert (
+            set_connection_output["properties"]["privateLinkServiceConnectionState"]["status"]
+            == "Rejected"
+        )
+        assert (
+            set_connection_output["properties"]["privateLinkServiceConnectionState"]["description"]
+            == random_desc_rejected
+        )
+
+        self.cmd(
+            "dt network private-endpoint connection delete -n {} -g {} --cn {} -y".format(
+                instance_name, self.rg, instance_connection_id
+            )
+        )
+
+        list_priv_endpoints = self.cmd(
+            "dt network private-endpoint connection list -n {} -g {}".format(
+                instance_name,
+                self.rg,
+            )
+        ).get_output_in_json()
+        assert len(list_priv_endpoints) == 0
+
+        # TODO clean-up optimization
+
+        self.cmd("network private-endpoint delete -n {} -g {} ".format(endpoint_name, self.rg))
+        self.cmd("network vnet delete -n {} -g {} ".format(vnet_name, self.rg))
diff --git a/pytest.ini.example b/pytest.ini.example
index b89a4be91..ede7b5ff2 100644
--- a/pytest.ini.example
+++ b/pytest.ini.example
@@ -19,6 +19,7 @@ env =
     azext_iot_teststorageuri=
     azext_iot_teststorageid=
     azext_iot_central_app_id=
+    azext_dt_region=
     azext_dt_ep_eventgrid_topic=
     azext_dt_ep_servicebus_namespace=
     azext_dt_ep_servicebus_policy=