PrivacyScanData/Azure/AzureStorage/tests/test_azure_6_list.py

from azure_storage.methods import client_prep, create_blob_service_client, extract_account_name, \
    extract_connection_string
from azure_storage.azure_list import AzureContainerList, AzureList, azure_search, cli, container_search
from unittest.mock import patch
import argparse
import pathlib
import pytest
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)
            self.test_path = os.path.abspath(os.path.dirname(__file__))
            self.file_path = os.path.join(self.test_path, 'files')
            self.output_file = os.path.join(self.file_path, 'search_outputs.txt')
            self.connection_string = extract_connection_string(passphrase=self.passphrase,
                                                               account_name=self.account_name)
            self.blob_service_client = create_blob_service_client(connect_str=self.connection_string)

    return Variables()


def read_output_file(output_file):
    contents = list()
    with open(output_file, 'r') as output:
        for line in output:
            contents.append(line.rstrip())
    return contents


def delete_output_file(output_file):
    os.remove(output_file)


def parse_file_outputs(check_file, contents):
    hit = False
    for entry in contents:
        if check_file in [os.path.basename(component) for component in entry.split('\t')]:
            hit = True
    return hit


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('expression',
                         ['000000container',
                          '0000\\d{2}container',
                          '0000\\d{2}\\D{9}',
                          '0000*',
                          '*container'])
def test_list_containers(variables, expression):
    AzureContainerList.list_containers(
        blob_service_client=variables.blob_service_client,
        expression=expression,
        print_container=False,
        output_file=variables.output_file)
    contents = read_output_file(output_file=variables.output_file)
    assert variables.container_name in contents
    delete_output_file(output_file=variables.output_file)


@pytest.mark.parametrize('expression',
                         ['000000container1',
                          '0000\\d{3}container',
                          '0000\\d{2}\\D{10}'
                          '100000*',
                          '*containers'])
def test_list_containers_invalid(variables, expression):
    with pytest.raises(FileNotFoundError):
        AzureContainerList.list_containers(
            blob_service_client=variables.blob_service_client,
            expression=expression,
            print_container=False,
            output_file=variables.output_file)
        read_output_file(output_file=variables.output_file)


def test_list_container_tilde(variables):
    path_obj = pathlib.Path(variables.file_path)
    path = f'~{os.sep}{path_obj.relative_to(pathlib.Path.home())}'
    list_containers = AzureContainerList(
        expression=variables.container_name,
        account_name=variables.account_name,
        output_file=os.path.join(path, 'search_outputs.txt'),
        passphrase=variables.passphrase
    )
    list_containers.main()
    contents = read_output_file(output_file=variables.output_file)
    assert variables.container_name in contents
    delete_output_file(output_file=variables.output_file)


def test_list_container_invalid_path(variables):
    output_file = os.path.join('/invalid', 'search_outputs.txt')
    if not os.environ.get('CIRCLECI'):
        with pytest.raises(SystemExit):
            AzureContainerList(
                expression=variables.container_name,
                account_name=variables.account_name,
                output_file=output_file,
                passphrase=variables.passphrase
            )


def test_list_container_directory_provided(variables):
    with pytest.raises(SystemExit):
        AzureContainerList(
            expression=variables.container_name,
            account_name=variables.account_name,
            output_file=variables.file_path,
            passphrase=variables.passphrase
        )


def test_list_container_no_output_file(variables):
    list_containers = AzureContainerList(
        expression=variables.container_name,
        account_name=variables.account_name,
        output_file=str(),
        passphrase=variables.passphrase
    )
    list_containers.main()
    assert not os.path.isfile(variables.output_file)


@patch('argparse.ArgumentParser.parse_args')
def test_list_container_integration(mock_args, variables):
    mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
                                                account_name=variables.account_name,
                                                expression='0*',
                                                verbosity='info',
                                                output_file=variables.output_file)
    arguments = cli()
    container_search(arguments)
    contents = read_output_file(output_file=variables.output_file)
    assert variables.container_name in contents
    delete_output_file(output_file=variables.output_file)


@pytest.mark.parametrize('container,expression,check_file',
                         [('000000container', 'file_1', 'file_1'),
                          ('000000container', 'file_1.gz', 'file_1.gz'),
                          ('000000container', 'container_integration/*_5/*.txt', 'nested_folder_test_1.txt'),
                          ('000000container', '*.txt', 'file_2.txt'),
                          ('000000container', '\\D{14}3/*_1.txt', 'nested_folder_test_1.txt'),
                          ('000000container', 'h\\D{2}/nes*', 'nested_folder_test_1.txt')])
def test_list_file(variables, container, expression, check_file):
    AzureList.list_files(container_client=variables.container_client,
                         expression=expression,
                         output_file=variables.output_file,
                         container_name=container)
    contents = read_output_file(output_file=variables.output_file)
    hit = parse_file_outputs(check_file=check_file,
                             contents=contents)
    assert hit
    delete_output_file(output_file=variables.output_file)


def test_list_file_tilde(variables):
    path_obj = pathlib.Path(variables.file_path)
    path = f'~{os.sep}{path_obj.relative_to(pathlib.Path.home())}'
    output_file = os.path.join(path, 'search_outputs.txt')
    check_file = 'triple_nested_file.txt'
    list_files = AzureList(
        container_name=variables.container_name,
        expression=check_file,
        account_name=variables.account_name,
        output_file=output_file,
        passphrase=variables.passphrase
    )
    list_files.main()
    contents = read_output_file(output_file=variables.output_file)
    hit = parse_file_outputs(check_file=check_file,
                             contents=contents)
    assert hit
    delete_output_file(output_file=variables.output_file)


def test_list_file_invalid_path(variables):
    output_file = os.path.join('/invalid', 'search_outputs.txt')
    if not os.environ.get('CIRCLECI'):
        with pytest.raises(SystemExit):
            AzureList(
                container_name=variables.container_name,
                expression=None,
                account_name=variables.account_name,
                output_file=output_file,
                passphrase=variables.passphrase
            )


def test_list_file_directory_provided(variables):
    with pytest.raises(SystemExit):
        AzureList(
            container_name=variables.container_name,
            expression=None,
            account_name=variables.account_name,
            output_file=variables.file_path,
            passphrase=variables.passphrase
        )


def test_list_file_no_output_file(variables):
    list_files = AzureList(
        container_name=variables.container_name,
        expression=None,
        account_name=variables.account_name,
        output_file=str(),
        passphrase=variables.passphrase
    )
    list_files.main()
    assert not os.path.isfile(variables.output_file)


@patch('argparse.ArgumentParser.parse_args')
def test_list_file_integration(mock_args, variables):
    mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
                                                account_name=variables.account_name,
                                                container_name=variables.container_name,
                                                expression='con*/c*/triple*file.txt',
                                                verbosity='info',
                                                output_file=variables.output_file)
    arguments = cli()
    azure_search(arguments)
    contents = read_output_file(output_file=variables.output_file)
    hit = False
    for entry in contents:
        if 'triple_nested_file.txt' in [os.path.basename(component) for component in entry.split('\t')]:
            hit = True
    assert hit
    delete_output_file(output_file=variables.output_file)


@patch('argparse.ArgumentParser.parse_args')
def test_list_file_container_expression_integration(mock_args, variables):
    mock_args.return_value = argparse.Namespace(passphrase=variables.passphrase,
                                                account_name=variables.account_name,
                                                container_name='0*',
                                                expression='nested_c*/double*file_1.txt',
                                                verbosity='info',
                                                output_file=variables.output_file)
    arguments = cli()
    azure_search(arguments)
    contents = read_output_file(output_file=variables.output_file)
    hit = False
    for entry in contents:
        if 'double_nested_file_1.txt' in [os.path.basename(component) for component in entry.split('\t')]:
            hit = True
    assert hit
    delete_output_file(output_file=variables.output_file)