Add live test case for blob copy scenario

src/azure-cli-testsdk/azure/cli/testsdk/base.py

@@ -31,30 +31,20 @@ class IntegrationTestBase(unittest.TestCase):
     def __init__(self, method_name):
         super(IntegrationTestBase, self).__init__(method_name)
         self.diagnose = os.environ.get(ENV_TEST_DIAGNOSE, None) == 'True'
-        self.skip_assert = os.environ.get(ENV_SKIP_ASSERT, None) == 'True'
 
-    def cmd(self, command, checks=None):
+    def cmd(self, command, checks=None, expect_failure=False):
         if self.diagnose:
             begin = datetime.datetime.now()
             print('\nExecuting command: {}'.format(command))
 
-        result = execute(command)
+        result = execute(command, expect_failure=expect_failure)
 
         if self.diagnose:
             duration = datetime.datetime.now() - begin
             print('\nCommand accomplished in {} s. Exit code {}.\n{}'.format(
                 duration.total_seconds(), result.exit_code, result.output))
 
-        if not checks:
-            checks = []
-        elif not isinstance(checks, list):
-            checks = [checks]
-
-        if not self.skip_assert:
-            for c in checks:
-                c(result)
-
-        return result
+        return result.assert_with_checks(checks)
 
     def create_random_name(self, prefix, length):  # for override pylint: disable=no-self-use
         return create_random_name(prefix, length)
@@ -229,6 +219,19 @@ def __init__(self, command, expect_failure=False, in_process=True):
             raise AssertionError('The command failed. Exit code: {}'.format(self.exit_code))
 
         self.json_value = None
+        self.skip_assert = os.environ.get(ENV_SKIP_ASSERT, None) == 'True'
+
+    def assert_with_checks(self, checks):
+        if not checks:
+            checks = []
+        elif not isinstance(checks, list):
+            checks = [checks]
+
+        if not self.skip_assert:
+            for c in checks:
+                c(self)
+
+        return self
 
     def get_output_in_json(self):
         if not self.json_value:

src/command_modules/azure-cli-storage/tests/storage_test_util.py

@@ -0,0 +1,29 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+
+class StorageScenarioMixin(object):
+    def get_account_key(self, group, name):
+        return self.cmd('storage account keys list -n {} -g {} --query "[0].value" -otsv'
+                        .format(name, group)).output
+
+    def get_account_info(self, group, name):
+        """Returns the storage account name and key in a tuple"""
+        return name, self.get_account_key(group, name)
+
+    def storage_cmd(self, cmd, account_info, *args):
+        cmd = cmd.format(*args)
+        cmd = '{} --account-name {} --account-key {}'.format(cmd, *account_info)
+        return self.cmd(cmd)
+
+    def storage_cmd_negative(self, cmd, account_info, *args):
+        cmd = cmd.format(*args)
+        cmd = '{} --account-name {} --account-key {}'.format(cmd, *account_info)
+        return self.cmd(cmd, expect_failure=True)
+
+    def create_container(self, account_info, prefix='cont', length=24):
+        container_name = self.create_random_name(prefix=prefix, length=length)
+        self.storage_cmd('storage container create -n {}', account_info, container_name)
+        return container_name

src/command_modules/azure-cli-storage/tests/test_storage_blob_copy_scenarios.py

@@ -0,0 +1,54 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from azure.cli.testsdk import (LiveTest, ResourceGroupPreparer, StorageAccountPreparer)
+from .storage_test_util import StorageScenarioMixin
+
+
+class StorageBlobCopyTests(StorageScenarioMixin, LiveTest):
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer(parameter_name='source_account')
+    @StorageAccountPreparer(parameter_name='target_account')
+    def test_storage_blob_copy_with_sas_and_snapshot(self, resource_group, source_account, target_account):
+        source_file = self.create_temp_file(16, full_random=True)
+        source_account_info = self.get_account_info(resource_group, source_account)
+        target_account_info = self.get_account_info(resource_group, target_account)
+
+        with open(source_file, 'rb') as f:
+            expect_content = f.read()
+
+        source_container = self.create_container(source_account_info)
+        target_container = self.create_container(target_account_info)
+
+        self.storage_cmd('storage blob upload -c {} -f {} -n src', source_account_info,
+                         source_container, source_file)
+
+        snapshot = self.storage_cmd('storage blob snapshot -c {} -n src', source_account_info,
+                                    source_container).get_output_in_json()['snapshot']
+
+        source_file = self.create_temp_file(24, full_random=True)
+        self.storage_cmd('storage blob upload -c {} -f {} -n src', source_account_info,
+                         source_container, source_file)
+
+        from datetime import datetime, timedelta
+        start = datetime.utcnow().strftime('%Y-%m-%dT%H:%MZ')
+        expiry = (datetime.utcnow() + timedelta(minutes=5)).strftime('%Y-%m-%dT%H:%MZ')
+        sas = self.storage_cmd('storage blob generate-sas -c {} -n src --permissions r --start {}'
+                               ' --expiry {}', source_account_info, source_container, start,
+                               expiry).output.strip()
+
+        self.storage_cmd('storage blob copy start -b dst -c {} --source-blob src --source-sas {} '
+                         '--source-account-name {} --source-container {} --source-snapshot {}',
+                         target_account_info, target_container, sas, source_account,
+                         source_container, snapshot)
+
+        target_file = self.create_temp_file(1)
+        self.storage_cmd('storage blob download -c {} -n dst -f {}', target_account_info,
+                         target_container, target_file)
+
+        with open(target_file, 'rb') as f:
+            actual_content = f.read()
+
+        self.assertEqual(expect_content, actual_content)

src/command_modules/azure-cli-storage/tests/test_storage_blob_live_scenarios.py

@@ -0,0 +1,64 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+from azure.cli.testsdk import (LiveTest, ResourceGroupPreparer, StorageAccountPreparer,
+                               JMESPathCheck)
+
+
+class StorageBlobUploadLiveTests(LiveTest):
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer()
+    def test_storage_blob_upload_256mb_file(self, resource_group, storage_account):
+        self.verify_blob_upload_and_download(resource_group, storage_account, 256 * 1024, 'block')
+
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer()
+    def test_storage_blob_upload_1G_file(self, resource_group, storage_account):
+        self.verify_blob_upload_and_download(resource_group, storage_account, 1024 * 1024, 'block')
+
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer()
+    def test_storage_blob_upload_2G_file(self, resource_group, storage_account):
+        self.verify_blob_upload_and_download(resource_group, storage_account, 2 * 1024 * 1024,
+                                             'block')
+
+    @ResourceGroupPreparer()
+    @StorageAccountPreparer()
+    def test_storage_blob_upload_10G_file(self, resource_group, storage_account):
+        self.verify_blob_upload_and_download(resource_group, storage_account, 10 * 1024 * 1024,
+                                             'block')
+
+    def verify_blob_upload_and_download(self, group, account, file_size_kb, blob_type):
+        container = self.create_random_name(prefix='cont', length=24)
+        local_dir = self.create_temp_dir()
+        local_file = self.create_temp_file(file_size_kb, full_random=True)
+        blob_name = self.create_random_name(prefix='blob', length=24)
+        account_key = self.cmd('storage account keys list -n {} -g {} --query "[0].value" -otsv'
+                               .format(account, group)).output
+
+        self.set_env('AZURE_STORAGE_ACCOUNT', account)
+        self.set_env('AZURE_STORAGE_KEY', account_key)
+
+        self.cmd('storage container create -n {}'.format(container))
+
+        self.cmd('storage blob exists -n {} -c {}'.format(blob_name, container),
+                 checks=JMESPathCheck('exists', False))
+
+        self.cmd('storage blob upload -c {} -f {} -n {} --type {}'
+                 .format(container, local_file, blob_name, blob_type))
+
+        self.cmd('storage blob exists -n {} -c {}'.format(blob_name, container),
+                 checks=JMESPathCheck('exists', True))
+
+        self.cmd('storage blob show -n {} -c {}'.format(blob_name, container),
+                 checks=JMESPathCheck('properties.contentLength', file_size_kb * 1024))
+
+        downloaded = os.path.join(local_dir, 'test.file')
+        self.cmd('storage blob download -n {} -c {} --file {}'
+                 .format(blob_name, container, downloaded))
+        self.assertTrue(os.path.isfile(downloaded), 'The file is not downloaded.')
+        self.assertEqual(file_size_kb * 1024, os.stat(downloaded).st_size,
+                         'The download file size is not right.')