from azure_storage.methods import extract_account_name, move_prep from azure_storage.azure_move import AzureContainerMove, AzureMove, cli, container_move, file_move, folder_move from unittest.mock import patch import argparse import pytest import azure import os @pytest.fixture(name='variables', scope='module') def setup(): class Variables: def __init__(self): # Extract the account name and connection string from the system keyring prior to running tests self.passphrase = 'AzureStorage' self.account_name = extract_account_name(passphrase=self.passphrase) self.container_name = '00000container' self.target_container = '000000container' self.storage_tier = 'Hot' return Variables() def run_move_prep(variables): variables.container_name, variables.target_container, variables.blob_service_client, \ variables.source_container_client, variables.target_container_client = move_prep( passphrase=variables.passphrase, account_name=variables.account_name, container_name=variables.container_name, target_container=variables.target_container) def test_move_prep(variables): run_move_prep(variables=variables) assert type(variables.source_container_client) is azure.storage.blob._container_client.ContainerClient def test_move_prep_target_exists(variables): run_move_prep(variables=variables) assert type(variables.target_container_client) is azure.storage.blob._container_client.ContainerClient @pytest.mark.parametrize('file_name,path', [('file_1.txt', ''), ('file_1', ''), ('file_1.gz', ''), ('file_2.txt', 'nested'), ('nested_folder/folder_test_1.txt', ''), ('nested_folder/nested_folder_2/nested_folder_test_1.txt', 'ABC'), ('nested_folder/nested_folder_2/nested_folder_test_1.txt', 'ABC/123'), ('nested/file_2.txt', 'nested_folder_2')]) def test_move_file(variables, file_name, path): AzureMove.move_file(source_container_client=variables.source_container_client, object_name=file_name, blob_service_client=variables.blob_service_client, container_name=variables.container_name, target_container=variables.target_container, path=path, storage_tier=variables.storage_tier) blobs = variables.target_container_client.list_blobs() assert os.path.join(path, os.path.basename(file_name)) in [blob.name for blob in blobs] @pytest.mark.parametrize('file_name,path', [('file_7.txt', ''), ('nonexistent/file_2.txt', 'nested_2'), ('nested/file_3.txt', 'nested_folder_2')]) def test_move_file_invalid(variables, file_name, path): with pytest.raises(SystemExit): AzureMove.move_file(source_container_client=variables.source_container_client, object_name=file_name, blob_service_client=variables.blob_service_client, container_name=variables.container_name, target_container=variables.target_container, path=path, storage_tier=variables.storage_tier) @pytest.mark.parametrize('file_name,path', [('nested_file_1.txt', ''), ('nested_folder/nested_file_2.txt', 'nested')]) def test_move_file_cool(variables, file_name, path): storage_tier = 'Cool' AzureMove.move_file(source_container_client=variables.source_container_client, object_name=file_name, blob_service_client=variables.blob_service_client, container_name=variables.container_name, target_container=variables.target_container, path=path, storage_tier=storage_tier) blobs = variables.target_container_client.list_blobs() for blob in blobs: if blob.name == os.path.join(path, file_name): assert blob.blob_tier == storage_tier @pytest.mark.parametrize('folder_name,path,check_file', [('single_nested', '', 'triple_nested_file.txt'), ('nested_folder/nested_folder_2', 'nested_folder_3', 'nested_folder_test_1.txt'), ('nested_folder/nested_folder_2', '', 'nested_folder_test_1.txt')]) def test_move_folder(variables, folder_name, path, check_file): AzureMove.move_folder(source_container_client=variables.source_container_client, object_name=folder_name, blob_service_client=variables.blob_service_client, container_name=variables.container_name, target_container=variables.target_container, path=path, storage_tier=variables.storage_tier, category='folder') blobs = variables.target_container_client.list_blobs() assert os.path.join(path, os.path.basename(check_file)) in [blob.name for blob in blobs] @pytest.mark.parametrize('folder_name,path,check_file', [('cool/nested_folder_2', '', 'nested_folder_test_1.txt'), ('cool_nested_folder_5', 'hot', 'nested_folder_test_1.txt')]) def test_move_folder_cool(variables, folder_name, path, check_file): AzureMove.move_folder(source_container_client=variables.source_container_client, object_name=folder_name, blob_service_client=variables.blob_service_client, container_name=variables.container_name, target_container=variables.target_container, path=path, storage_tier=variables.storage_tier, category='folder') blobs = variables.target_container_client.list_blobs() for blob in blobs: if blob.name == os.path.join(path, check_file): assert blob.blob_tier == variables.storage_tier @pytest.mark.parametrize('folder_name,path', [('nested_folder_6', ''), ('nested_folder_2/nested_folder_2', 'nested_folder'), ('nested_folder/nested_folder_3', '')]) def test_move_folder_invalid(variables, folder_name, path): with pytest.raises(SystemExit): AzureMove.move_folder(source_container_client=variables.source_container_client, object_name=folder_name, blob_service_client=variables.blob_service_client, container_name=variables.container_name, target_container=variables.target_container, path=path, storage_tier=variables.storage_tier, category='folder') def test_move_folder_invalid_category(variables): with pytest.raises(SystemExit): move_folder = AzureMove(object_name='cool/nested_folder_2', container_name=variables.container_name, account_name=variables.account_name, passphrase=variables.passphrase, target_container=variables.target_container, path=None, storage_tier=variables.storage_tier, category='container') move_folder.main() def test_move_container(variables): path = 'nested_container' AzureContainerMove.move_container(source_container_client=variables.source_container_client, blob_service_client=variables.blob_service_client, container_name=variables.container_name, target_container=variables.target_container, path=path, storage_tier=variables.storage_tier) blobs = variables.target_container_client.list_blobs() assert os.path.join(path, 'file_2.txt') in [blob.name for blob in blobs] @patch('argparse.ArgumentParser.parse_args') def test_move_file_integration(mock_args, variables): file_name = 'file_1.txt' reset_path = 'file_integration' mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase, account_name=variables.account_name, container_name=variables.container_name, target_container=variables.target_container, reset_path=reset_path, verbosity='info', file=file_name, storage_tier=variables.storage_tier) arguments = cli() file_move(arguments) blobs = variables.target_container_client.list_blobs() assert os.path.join(reset_path, file_name) in [blob.name for blob in blobs] @patch('argparse.ArgumentParser.parse_args') def test_move_file_integration_cool(mock_args, variables): file_name = 'nested_file_1.txt' reset_path = 'cool_file_integration' storage_tier = 'Cool' mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase, account_name=variables.account_name, container_name=variables.container_name, target_container=variables.target_container, reset_path=reset_path, verbosity='info', file=file_name, storage_tier=storage_tier) arguments = cli() file_move(arguments) blobs = variables.target_container_client.list_blobs() for blob in blobs: if blob.name == os.path.join(reset_path, file_name): assert blob.blob_tier == storage_tier @patch('argparse.ArgumentParser.parse_args') def test_move_folder_integration(mock_args, variables): folder_name = 'nested_folder_5' reset_path = 'folder_integration' mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase, account_name=variables.account_name, container_name=variables.container_name, target_container=variables.target_container, reset_path=reset_path, verbosity='info', folder=folder_name, storage_tier=variables.storage_tier) arguments = cli() folder_move(arguments) blobs = variables.target_container_client.list_blobs() assert os.path.join(reset_path, 'nested_folder_test_1.txt') in [blob.name for blob in blobs] @patch('argparse.ArgumentParser.parse_args') def test_move_container_integration(mock_args, variables): reset_path = 'container_integration' mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase, account_name=variables.account_name, container_name=variables.container_name, target_container=variables.target_container, reset_path=reset_path, verbosity='info', storage_tier=variables.storage_tier) arguments = cli() container_move(arguments) blobs = variables.target_container_client.list_blobs() assert os.path.join(reset_path, 'double_nested_file_1.txt') in [blob.name for blob in blobs]