[Storage] allow deletion of containers with immutability policy after…

… service bug fix (#7569)

* delete container to custom command

* finished up commands

* added test

* history and pep8

* linter, static

* new recording after scrubbing fix

* comment typo

src/command_modules/azure-cli-storage/HISTORY.rst

@@ -6,6 +6,7 @@ Release History
 2.2.3
 +++++
 * Fix `az storage cors list` output formatting, all items show correct "Service" key
+* `--bypass-immutability-policy` parameter for immutability-policy blocked container deletion
 
 2.2.2
 +++++

src/command_modules/azure-cli-storage/azure/cli/command_modules/storage/_help.py

@@ -100,6 +100,13 @@
           text: az storage container create -n MyStorageContainer --fail-on-exist
 """
 
+helps['storage container delete'] = """
+    type: command
+    short-summary: Marks the specified container for deletion.
+    long-summary: >
+        The container and any blobs contained within it are later deleted during garbage collection.
+"""
+
 helps['storage account list'] = """
     type: command
     short-summary: List storage accounts.

src/command_modules/azure-cli-storage/azure/cli/command_modules/storage/_params.py

@@ -370,6 +370,14 @@ def load_arguments(self, _):  # pylint: disable=too-many-locals, too-many-statem
 
     with self.argument_context('storage container delete') as c:
         c.argument('fail_not_exist', help='Throw an exception if the container does not exist.')
+        c.argument('bypass_immutability_policy', action='store_true', help='Bypasses upcoming service behavior that '
+                   'will block a container from being deleted if it has a immutability-policy. Specifying this will '
+                   'ignore arguments aside from those used to identify the container ("--name", "--account-name").')
+        c.argument('lease_id', help="If specified, delete_container only succeeds if the container's lease is active "
+                                    "and matches this ID. Required if the container has an active lease.")
+        c.ignore('processed_resource_group')
+        c.ignore('processed_account_name')
+        c.ignore('mgmt_client')
 
     with self.argument_context('storage container exists') as c:
         c.ignore('blob_name', 'snapshot')

src/command_modules/azure-cli-storage/azure/cli/command_modules/storage/_validators.py

@@ -713,6 +713,20 @@ def _update_progress(current, total):
     del namespace.no_progress
 
 
+def process_container_delete_parameters(cmd, namespace):
+    """Process the parameters for storage container delete command"""
+    # check whether to use mgmt or data-plane
+    if namespace.bypass_immutability_policy:
+        # use management-plane
+        namespace.processed_account_name = namespace.account_name
+        namespace.processed_resource_group, namespace.mgmt_client = _query_account_rg(
+            cmd.cli_ctx, namespace.account_name)
+        del namespace.auth_mode
+    else:
+        # use data-plane, like before
+        validate_client_parameters(cmd, namespace)
+
+
 def process_blob_download_batch_parameters(cmd, namespace):
     """Process the parameters for storage blob download command"""
     import os

src/command_modules/azure-cli-storage/azure/cli/command_modules/storage/commands.py

@@ -165,18 +165,19 @@ def get_custom_sdk(custom_module, client_factory, resource_type=ResourceType.DAT
         g.storage_command_oauth('show', 'get_blob_service_properties', exception_handler=show_exception_handler)
 
     with self.command_group('storage container', command_type=block_blob_sdk,
-                            custom_command_type=get_custom_sdk('acl', blob_data_service_factory)) as g:
+                            custom_command_type=get_custom_sdk('blob', blob_data_service_factory)) as g:
         from azure.cli.command_modules.storage._transformers import (transform_storage_list_output,
                                                                      transform_container_permission_output,
                                                                      transform_acl_list_output)
         from azure.cli.command_modules.storage._format import (transform_container_list, transform_boolean_for_table,
                                                                transform_container_show)
+        from ._validators import process_container_delete_parameters
 
         g.storage_command_oauth('list', 'list_containers', transform=transform_storage_list_output,
                                 table_transformer=transform_container_list)
-        g.storage_command_oauth('delete', 'delete_container',
-                                transform=create_boolean_result_output_transformer('deleted'),
-                                table_transformer=transform_boolean_for_table)
+        g.storage_custom_command_oauth('delete', 'delete_container', validator=process_container_delete_parameters,
+                                       transform=create_boolean_result_output_transformer('deleted'),
+                                       table_transformer=transform_boolean_for_table)
         g.storage_command_oauth('show', 'get_container_properties', table_transformer=transform_container_show,
                                 exception_handler=show_exception_handler)
         g.storage_command_oauth('create', 'create_container',
@@ -196,6 +197,8 @@ def get_custom_sdk(custom_module, client_factory, resource_type=ResourceType.DAT
         g.storage_command_oauth('lease change', 'change_container_lease')
         g.storage_command_oauth('lease break', 'break_container_lease')
 
+    with self.command_group('storage container', command_type=block_blob_sdk,
+                            custom_command_type=get_custom_sdk('acl', blob_data_service_factory)) as g:
         g.storage_custom_command_oauth('policy create', 'create_acl_policy')
         g.storage_custom_command_oauth('policy delete', 'delete_acl_policy')
         g.storage_custom_command_oauth('policy update', 'set_acl_policy', min_api='2017-04-17')

src/command_modules/azure-cli-storage/azure/cli/command_modules/storage/operations/blob.py

@@ -8,6 +8,7 @@
 import os
 from knack.log import get_logger
 
+
 from azure.cli.command_modules.storage.util import (create_blob_service_from_storage_client,
                                                     create_file_share_from_storage_client,
                                                     create_short_lived_share_sas,
@@ -18,6 +19,16 @@
 from azure.cli.command_modules.storage.url_quote_util import encode_for_url, make_encoded_file_url_and_params
 
 
+def delete_container(client, container_name, fail_not_exist=False, lease_id=None, if_modified_since=None,
+                     if_unmodified_since=None, timeout=None, bypass_immutability_policy=False,
+                     processed_resource_group=None, processed_account_name=None, mgmt_client=None):
+    if bypass_immutability_policy:
+        return mgmt_client.blob_containers.delete(processed_resource_group, processed_account_name, container_name)
+    return client.delete_container(
+        container_name, fail_not_exist=fail_not_exist, lease_id=lease_id, if_modified_since=if_modified_since,
+        if_unmodified_since=if_unmodified_since, timeout=timeout)
+
+
 def set_blob_tier(client, container_name, blob_name, tier, blob_type='block', timeout=None):
     if blob_type == 'block':
         return client.set_standard_blob_tier(container_name=container_name, blob_name=blob_name,

src/command_modules/azure-cli-storage/azure/cli/command_modules/storage/tests/latest/test_storage_container_immutability_policy.py

@@ -54,3 +54,5 @@ def test_immutability_policy(self, resource_group):
                  '--account-name {} -c {} -g {} --if-match {}'.format(
                      storage_account, container_name, resource_group, repr(policy_etag)), checks=[
                          JMESPathCheck('immutabilityPeriodSinceCreationInDays', 2)])
+
+        self.cmd('az storage container delete --account-name {} -n {}'.format(storage_account, container_name))