[Storage] Support Version level Worm (VLW)

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

@@ -1179,6 +1179,18 @@
     text: az storage container-rm list --storage-account myaccount --include-deleted
 """
 
+helps['storage container-rm migrate-vlw'] = """
+type: command
+short-summary: Migrate a blob container from container level WORM to object level immutability enabled container.
+examples:
+  - name: Migrate a blob container from container level WORM to object level immutability enabled container.
+    text: az storage container-rm migrate-vlw -n mycontainer --storage-account myaccount -g myresourcegroup
+  - name: Migrate a blob container from container level WORM to object level immutability enabled container without waiting.
+    text: |
+        az storage container-rm migrate-vlw -n mycontainer --storage-account myaccount -g myresourcegroup --no-wait
+        az storage container-rm show -n mycontainer --storage-account myaccount -g myresourcegroup  --query immutableStorageWithVersioning.migrationState
+"""
+
 helps['storage container-rm show'] = """
 type: command
 short-summary: Show the properties for a specified container.

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

@@ -1248,6 +1248,10 @@ def load_arguments(self, _):  # pylint: disable=too-many-locals, too-many-statem
 
     with self.argument_context('storage container-rm create', resource_type=ResourceType.MGMT_STORAGE) as c:
         c.argument('fail_on_exist', help='Throw an exception if the container already exists.')
+        c.argument('enable_vlw', arg_type=get_three_state_flag(), min_api='2021-01-01', is_preview=True,
+                   help='The object level immutability property of the container. The property is immutable and can '
+                   'only be set to true at the container creation time. Existing containers must undergo a migration '
+                   'process.')
 
     for item in ['create', 'update']:
         with self.argument_context('storage container-rm {}'.format(item),

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

@@ -503,7 +503,7 @@ def get_custom_sdk(custom_module, client_factory, resource_type=ResourceType.DAT
     with self.command_group('storage container-rm', command_type=blob_container_mgmt_sdk,
                             custom_command_type=get_custom_sdk('blob', cf_blob_container_mgmt,
                                                                resource_type=ResourceType.MGMT_STORAGE),
-                            resource_type=ResourceType.MGMT_STORAGE, min_api='2018-02-01', is_preview=True) as g:
+                            resource_type=ResourceType.MGMT_STORAGE, min_api='2018-02-01') as g:
         g.custom_command('create', 'create_container_rm')
         g.command('delete', 'delete', confirmation=True)
         g.generic_update_command('update', setter_name='update', max_api='2019-04-01')
@@ -513,6 +513,7 @@ def get_custom_sdk(custom_module, client_factory, resource_type=ResourceType.DAT
         g.custom_command('exists', 'container_rm_exists', transform=create_boolean_result_output_transformer('exists'),
                          table_transformer=transform_boolean_for_table)
         g.show_command('show', 'get')
+        g.command('migrate-vlw', 'begin_object_level_worm', supports_no_wait=True, is_preview=True)
 
     file_sdk = CliCommandType(
         operations_tmpl='azure.multiapi.storage.file.fileservice#FileService.{}',

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

@@ -57,7 +57,7 @@ def extend_immutability_policy(cmd, client, container_name, account_name, if_mat
 
 def create_container_rm(cmd, client, container_name, resource_group_name, account_name,
                         metadata=None, public_access=None, fail_on_exist=False,
-                        default_encryption_scope=None, deny_encryption_scope_override=None):
+                        default_encryption_scope=None, deny_encryption_scope_override=None, enable_vlw=None):
     if fail_on_exist and container_rm_exists(client, resource_group_name=resource_group_name,
                                              account_name=account_name, container_name=container_name):
         raise CLIError('The specified container already exists.')
@@ -68,6 +68,10 @@ def create_container_rm(cmd, client, container_name, resource_group_name, accoun
                                        default_encryption_scope=default_encryption_scope,
                                        deny_encryption_scope_override=deny_encryption_scope_override,
                                        metadata=metadata)
+        if enable_vlw is not None:
+            ImmutableStorageWithVersioning = cmd.get_models('ImmutableStorageWithVersioning',
+                                                            resource_type=ResourceType.MGMT_STORAGE)
+            blob_container.immutable_storage_with_versioning = ImmutableStorageWithVersioning(enabled=enable_vlw)
         return client.create(resource_group_name=resource_group_name, account_name=account_name,
                              container_name=container_name, blob_container=blob_container)
     return client.create(resource_group_name=resource_group_name, account_name=account_name,

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

@@ -785,6 +785,48 @@ def test_storage_blob_copy_rehydrate_priority(self, resource_group, storage_acco
             .assert_with_checks(JMESPathCheck('properties.blobTier', 'Archive'),
                                 JMESPathCheck('properties.rehydrationStatus', 'rehydrate-pending-to-cool'))
 
+    @AllowLargeResponse()
+    @ResourceGroupPreparer(name_prefix='clitest')
+    @StorageAccountPreparer(kind='StorageV2', name_prefix='clitest', location='centraluseuap')
+    def test_storage_container_vlm_scenarios(self, resource_group, storage_account):
+        self.kwargs.update({
+            'container1': self.create_random_name(prefix='con1', length=10),
+            'container2': self.create_random_name(prefix='con2', length=10)
+        })
+        self.cmd('storage account blob-service-properties update -n {sa} -g {rg} --enable-versioning ',
+                 checks={
+                     JMESPathCheck('isVersioningEnabled', True)
+                 })
+        # Enable vlm when creation
+        self.cmd('storage container-rm create -n {container1} --storage-account {sa} -g {rg} --enable-vlw',
+                 checks={
+                     JMESPathCheck('name', self.kwargs['container1']),
+                     JMESPathCheck('immutableStorageWithVersioning.enabled', True),
+                     JMESPathCheck('immutableStorageWithVersioning.migrationState', None)})
+        self.cmd('storage container-rm show -n {container1} --storage-account {sa} -g {rg}',
+                 checks={
+                     JMESPathCheck('name', self.kwargs['container1']),
+                     JMESPathCheck('immutableStorageWithVersioning.enabled', True),
+                     JMESPathCheck('immutableStorageWithVersioning.migrationState', None)})
+
+        # Enable vlm for containers with immutability policy
+        self.cmd('storage container-rm create -n {container2} --storage-account {sa} -g {rg}',
+                 checks={
+                     JMESPathCheck('name', self.kwargs['container2']),
+                     JMESPathCheck('immutableStorageWithVersioning.enabled', None)})
+
+        self.cmd('storage container immutability-policy create -c {container2} --account-name {sa} -w --period 1',
+                 checks={
+                     JMESPathCheck('name', self.kwargs['container2']),
+                     JMESPathCheck('immutabilityPeriodSinceCreationInDays', 1)})
+
+        self.cmd('storage container-rm migrate-vlw -n {container2} --storage-account {sa} -g {rg} --no-wait')
+        self.cmd('storage container-rm show -n {container2} --storage-account {sa} -g {rg}',
+                 checks={
+                     JMESPathCheck('name', self.kwargs['container2']),
+                     JMESPathCheck('immutableStorageWithVersioning.enabled', False),
+                     JMESPathCheck('immutableStorageWithVersioning.migrationState', 'InProgress')})
+
 
 if __name__ == '__main__':
     unittest.main()