123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186 |
- from azure_storage.methods import client_prep, delete_container, delete_file, delete_folder, extract_account_name
- from azure_storage.azure_delete import AzureDelete, cli, container_delete, file_delete, \
- folder_delete
- 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):
- self.passphrase = 'AzureStorage'
- self.container_name = '000000container'
- self.account_name = extract_account_name(passphrase=self.passphrase)
- return Variables()
- def test_client_prep(variables):
- variables.container_name, variables.connect_str, variables.blob_service_client, variables.container_client = \
- client_prep(container_name=variables.container_name,
- passphrase=variables.passphrase,
- account_name=variables.account_name)
- assert variables.connect_str.startswith('DefaultEndpointsProtocol')
- @pytest.mark.parametrize('file_name',
- ['file_1.txt',
- 'container_integration/file_2.txt',
- 'nested_container/nested_folder/nested_folder_2/nested_folder_test_1.txt',
- 'ABC/123/nested_folder_test_1.txt'])
- def test_delete_file(variables, file_name):
- delete_file(container_client=variables.container_client,
- object_name=file_name,
- blob_service_client=variables.blob_service_client,
- container_name=variables.container_name)
- blobs = variables.container_client.list_blobs()
- assert file_name not in [blob.name for blob in blobs]
- @pytest.mark.parametrize('file_name',
- ['file_3.txt',
- 'container_integration/file_2.txt',
- 'nested_container/nested_folder/nested_folder_2/nested_folder_test_1.txt',
- 'ABC/123/nested_folder_test_1.txt'])
- def test_delete_file_missing(variables, file_name):
- with pytest.raises(SystemExit):
- delete_file(container_client=variables.container_client,
- object_name=file_name,
- blob_service_client=variables.blob_service_client,
- container_name=variables.container_name)
- def test_delete_file_invalid_category(variables):
- with pytest.raises(SystemExit):
- del_file = AzureDelete(object_name='file_1.txt',
- container_name=variables.container_name,
- account_name=variables.account_name,
- passphrase=variables.passphrase,
- retention_time=8,
- category='container')
- del_file.main()
- @patch('argparse.ArgumentParser.parse_args')
- def test_delete_file_integration(mock_args, variables):
- file_name = 'nested/file_2.txt'
- mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
- account_name=variables.account_name,
- container_name=variables.container_name,
- verbosity='info',
- file=file_name,
- retention_time=1)
- arguments = cli()
- file_delete(arguments)
- blobs = variables.container_client.list_blobs()
- assert os.path.basename(file_name) not in [blob.name for blob in blobs]
- @pytest.mark.parametrize('retention_time',
- [0,
- 1000])
- @patch('argparse.ArgumentParser.parse_args')
- def test_delete_file_integration_invalid_retention_time(mock_args, variables, retention_time):
- file_name = 'nested/file_2.txt'
- with pytest.raises(SystemExit):
- mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
- account_name=variables.account_name,
- container_name=variables.container_name,
- verbosity='info',
- file=file_name,
- retention_time=retention_time)
- arguments = cli()
- file_delete(arguments)
- @patch('argparse.ArgumentParser.parse_args')
- def test_delete_file_integration_missing(mock_args, variables):
- file_name = 'nested/file_2.txt'
- with pytest.raises(SystemExit):
- mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
- account_name=variables.account_name,
- container_name=variables.container_name,
- verbosity='info',
- file=file_name,
- retention_time=1)
- arguments = cli()
- file_delete(arguments)
- @pytest.mark.parametrize('folder_name,check_file',
- [('container_integration/', 'nested_folder_test_1.txt'),
- ('nested_container/nested_folder/', 'nested_file_2.txt'),
- ('ABC/', 'nested_folder_test_1.txt')])
- def test_delete_folder(variables, folder_name, check_file):
- delete_folder(container_client=variables.container_client,
- object_name=folder_name,
- blob_service_client=variables.blob_service_client,
- container_name=variables.container_name,
- account_name=variables.account_name)
- blobs = variables.container_client.list_blobs()
- assert os.path.join(folder_name, check_file) not in [blob.name for blob in blobs]
- @pytest.mark.parametrize('folder_name,check_file',
- [('container_integration/', 'nested_folder_test_1.txt'),
- ('nested_container/nested_folder/', 'nested_file_2.txt'),
- ('ABC/', 'nested_folder_test_1.txt')])
- def test_delete_folder_missing(variables, folder_name, check_file):
- with pytest.raises(SystemExit):
- delete_folder(container_client=variables.container_client,
- object_name=folder_name,
- blob_service_client=variables.blob_service_client,
- container_name=variables.container_name,
- account_name=variables.account_name)
- @patch('argparse.ArgumentParser.parse_args')
- def test_delete_folder_integration(mock_args, variables):
- folder_name = 'nested_folder_3'
- mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
- account_name=variables.account_name,
- container_name=variables.container_name,
- verbosity='info',
- folder=folder_name,
- retention_time=1)
- arguments = cli()
- folder_delete(arguments)
- blobs = variables.container_client.list_blobs()
- assert os.path.join(folder_name, 'nested_folder_test_1.txt') not in [blob.name for blob in blobs]
- @patch('argparse.ArgumentParser.parse_args')
- def test_delete_folder_integration_missing(mock_args, variables):
- folder_name = 'nested_folder_3'
- with pytest.raises(SystemExit):
- mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
- account_name=variables.account_name,
- container_name=variables.container_name,
- verbosity='info',
- folder=folder_name,
- retention_time=1)
- arguments = cli()
- folder_delete(arguments)
- def test_delete_container_missing(variables):
- with pytest.raises(SystemExit):
- delete_container(blob_service_client=variables.blob_service_client,
- container_name='000000000container',
- account_name=variables.account_name)
- @patch('argparse.ArgumentParser.parse_args')
- def test_delete_container_integration(mock_args, variables):
- mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
- account_name=variables.account_name,
- container_name=variables.container_name,
- verbosity='info')
- arguments = cli()
- container_delete(arguments)
- with pytest.raises(azure.core.exceptions.ResourceExistsError):
- variables.blob_service_client.create_container(variables.container_name)
|