[Network] Support private link for managed disks (Azure#14707)

* support private link for disks

* fix typos

* adjust error message

* code refining

Author: houk-ms

src/azure-cli/azure/cli/command_modules/network/private_link_resource_and_endpoint_connections/custom.py

@@ -31,9 +31,10 @@ def register_providers():
     _register_one_provider('Microsoft.EventGrid/domains', '2020-04-01-preview', True)
     _register_one_provider('Microsoft.SignalRService/signalr', '2020-05-01', False)
     _register_one_provider('Microsoft.Network/applicationGateways', '2020-05-01', True)
+    _register_one_provider('Microsoft.Compute/diskAccesses', '2020-05-01', False, support_connection_operation=False)
 
 
-def _register_one_provider(provider, api_version, support_list_or_not, resource_get_api_version=None):
+def _register_one_provider(provider, api_version, support_list_or_not, resource_get_api_version=None, support_connection_operation=True):  # pylint: disable=line-too-long
     """
     :param provider: namespace + type.
     :param api_version: API version for private link scenarios.
@@ -43,12 +44,20 @@ def _register_one_provider(provider, api_version, support_list_or_not, resource_
     general_client_settings = {
         "api_version": api_version,
         "support_list_or_not": support_list_or_not,
-        "resource_get_api_version": resource_get_api_version
+        "resource_get_api_version": resource_get_api_version,
+        "support_connection_operation": support_connection_operation
     }
 
     TYPE_CLIENT_MAPPING[provider] = general_client_settings
 
 
+def _check_connection_operation_support(rp_mapping, resource_provider):
+    if resource_provider in rp_mapping \
+       and isinstance(rp_mapping[resource_provider], dict) \
+       and not rp_mapping[resource_provider]['support_connection_operation']:
+        raise CLIError("Resource provider {} currently does not support this operation".format(resource_provider))
+
+
 def _get_client(rp_mapping, resource_provider):
     for key, value in rp_mapping.items():
         if str.lower(key) == str.lower(resource_provider):
@@ -68,6 +77,7 @@ def list_private_link_resource(cmd, resource_group_name, name, resource_provider
 
 def approve_private_endpoint_connection(cmd, resource_group_name, resource_name, resource_provider,
                                         name, approval_description=None):
+    _check_connection_operation_support(TYPE_CLIENT_MAPPING, resource_provider)
     client = _get_client(TYPE_CLIENT_MAPPING, resource_provider)
     return client.approve_private_endpoint_connection(cmd, resource_group_name,
                                                       resource_name, name,
@@ -76,18 +86,21 @@ def approve_private_endpoint_connection(cmd, resource_group_name, resource_name,
 
 def reject_private_endpoint_connection(cmd, resource_group_name, resource_name, resource_provider,
                                        name, rejection_description=None):
+    _check_connection_operation_support(TYPE_CLIENT_MAPPING, resource_provider)
     client = _get_client(TYPE_CLIENT_MAPPING, resource_provider)
     return client.reject_private_endpoint_connection(cmd, resource_group_name,
                                                      resource_name, name,
                                                      rejection_description)
 
 
 def remove_private_endpoint_connection(cmd, resource_group_name, resource_name, resource_provider, name):
+    _check_connection_operation_support(TYPE_CLIENT_MAPPING, resource_provider)
     client = _get_client(TYPE_CLIENT_MAPPING, resource_provider)
     return client.remove_private_endpoint_connection(cmd, resource_group_name, resource_name, name)
 
 
 def show_private_endpoint_connection(cmd, resource_group_name, resource_name, resource_provider, name):
+    _check_connection_operation_support(TYPE_CLIENT_MAPPING, resource_provider)
     client = _get_client(TYPE_CLIENT_MAPPING, resource_provider)
     return client.show_private_endpoint_connection(cmd, resource_group_name, resource_name, name)
 

src/azure-cli/azure/cli/command_modules/network/tests/latest/recordings/test_disk_access_private_endpoint.yaml

@@ -1342,5 +1342,54 @@ def test_manage_appgw_private_endpoint(self, resource_group):
         # self.cmd('network application-gateway private-link list -g {rg} --gateway-name {appgw} ')
 
 
+class NetworkPrivateLinkDiskAccessScenarioTest(ScenarioTest):
+    @ResourceGroupPreparer(name_prefix='test_disk_access_private_endpoint_')
+    def test_disk_access_private_endpoint(self, resource_group):
+        self.kwargs.update({
+            'loc': 'centraluseuap',
+            'disk_access': 'disk_access_name',
+            'pe_vnet': 'private_endpoint_vnet',
+            'pe_subnet': 'private_endpoint_subnet',
+            'pe_name': 'private_endpoint_name',
+            'pe_connection': 'private_connection_name'
+        })
+
+        # Create disk access for private endpoint
+        disk_access_output = self.cmd('disk-access create -g {rg} -l {loc} -n {disk_access}').get_output_in_json()
+        self.kwargs.update({
+            'disk_access_id': disk_access_output['id']
+        })
+
+        # Check private link resource
+        self.cmd('network private-link-resource list -g {rg} -n {disk_access} --type Microsoft.Compute/diskAccesses',
+                 checks=[
+                     self.check('length(@)', 1),
+                     self.check('@[0].name', 'disks')
+                 ])
+
+        # Prepare the vnet to be connected to
+        self.cmd('network vnet create -g {rg} -n {pe_vnet} --subnet-name {pe_subnet}')
+        # Enable private endpoint on a vnet would require --disable-private-endpoint-network-policies=true
+        self.cmd('network vnet subnet update -g {rg} -n {pe_subnet} '
+                 '--vnet-name {pe_vnet} '
+                 '--disable-private-endpoint-network-policies true')
+
+        # Create a private endpoint connection for the disk access object
+        pe_output = self.cmd('network private-endpoint create -g {rg} -n {pe_name} --vnet-name {pe_vnet} '
+                             '--subnet {pe_subnet} --private-connection-resource-id {disk_access_id} '
+                             '--group-ids disks --connection-name {pe_connection}').get_output_in_json()
+        self.kwargs.update({
+            'pe_id': pe_output['id']
+        })
+
+        # Check the auto-approve of the private endpoint connection
+        self.cmd('network private-endpoint-connection list -g {rg} -n {disk_access} --type Microsoft.Compute/diskAccesses',
+                 checks=[
+                     self.check('length(@)', 1),
+                     self.check('@[0].properties.privateEndpoint.id', '{pe_id}'),
+                     self.check('@[0].properties.privateLinkServiceConnectionState.status', 'Approved')
+                 ])
+
+
 if __name__ == '__main__':
     unittest.main()