Index: /branches/ipp-132_automate_jira_conf_backups/backups/backup_atlassian_applications.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/backup_atlassian_applications.py (revision 40897) +++ /branches/ipp-132_automate_jira_conf_backups/backups/backup_atlassian_applications.py (revision 40898) @@ -9,20 +9,8 @@ import sys -# note that this replies on sym link to the 'trunk/tools/utils/python' folder existing -# in this directory with the name 'utils'. This is not an elegant solution, but -# a restructure would be needed -# Ideally, the utils should be added to the path. -import subprocess_utils as sub_utils - +import backups.utils.subprocess_utils as sub_utils +from backups.utils.errors import ConfigError, SubprocessError DEFAULT_CONFIG_FILE = path.join(path.dirname(__file__), 'atlassian_backups.config') - - -class ConfigError(Exception): - """Error in loading the config file""" - - def __init__(self, config_item, message): - self.config_item = config_item - self.message = message @@ -284,5 +272,5 @@ decoded_stderr = stde.decode() if decoded_stderr.casefold().find('error'.casefold()) > -1: - raise sub_utils.SubprocessError() + raise SubprocessError() # move the dump file to the correct locations Index: /branches/ipp-132_automate_jira_conf_backups/backups/backup_atlassian_applications_test.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/backup_atlassian_applications_test.py (revision 40897) +++ /branches/ipp-132_automate_jira_conf_backups/backups/backup_atlassian_applications_test.py (revision 40898) @@ -1,3 +1,3 @@ -import backup_atlassian_applications as baa +import backups.backup_atlassian_applications as baa import configparser import datetime @@ -7,8 +7,8 @@ from pathlib import Path -import full_atlassian_backup_test as fab_test -import subprocess_utils as sub_utils - -from backup_atlassian_applications import AtlassianBackups +import backups.full_atlassian_backup_test as fab_test +import backups.utils.subprocess_utils as sub_utils + +from backups.backup_atlassian_applications import AtlassianBackups TEST_DEFAULT_CONFIG_FILE = os.path.join(os.path.dirname(__file__), './testing/test.config') Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/database/database_tool.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/database/database_tool.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/database/database_tool.py (revision 40898) @@ -0,0 +1,37 @@ +import mysql.connector +from mysql.connector import Error + + +# To install this, you'll need: +# pip install mysql-connector-python + +def connection(): + + try: + connection_config_dict = { + 'user': 'pynative', + 'password': 'pynative@123', + 'host': '127.0.0.1', + 'database': 'Electronics', + 'raise_on_warnings': True, + 'use_pure': False, + 'autocommit': True, + 'pool_size': 5 + } + connection = mysql.connector.connect(**connection_config_dict) + + if connection.is_connected(): + db_Info = connection.get_server_info() + print("Connected to MySQL Server version ", db_Info) + cursor = connection.cursor() + cursor.execute("select database();") + record = cursor.fetchone() + print("Your connected to database: ", record) + + except Error as e: + print("Error while connecting to MySQL", e) + finally: + if (connection.is_connected()): + cursor.close() + connection.close() + print("MySQL connection is closed") Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/database/database_tool_test.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/database/database_tool_test.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/database/database_tool_test.py (revision 40898) @@ -0,0 +1,9 @@ +import backups.utils.database.database_tool as dbt + + +class TestMySQLConnections(object): + + def test_connection(self): + # dbt.connection() + + assert True is True Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/errors.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/errors.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/errors.py (revision 40898) @@ -0,0 +1,16 @@ + + +class ConfigError(Exception): + """Error in loading the config file""" + + def __init__(self, config_item, message): + self.config_item = config_item + self.message = message + + +class ValidationError(Exception): + """Validation errors in utils functions. Raised when arguments are not valid""" + + +class SubprocessError(Exception): + """Errors in Popen stderr subprocess in subprocess_utils""" Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/generic_utils.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/generic_utils.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/generic_utils.py (revision 40898) @@ -0,0 +1,67 @@ +import os.path as path +import shutil + +from distutils.dir_util import copy_tree +from backups.utils.errors import ValidationError + + +def copy_to_dir(source_file: str, destination_path: str, verbose: bool=False): + """ Copies a single file into the destination directory. If a directory is + given as a source then all the contents of the directory will be copied (the + directory itself will not be). +""" + + if source_file is None or destination_path is None: + raise ValidationError("Error: source for copy cannot be None type") + if destination_path is None: + raise ValidationError("Error: copy destination cannot be None type") + if not path.exists(source_file): + raise ValidationError(f"Error: Copy Failure: source does not exist: {source_file}") + if not path.exists(destination_path) or not path.isdir(destination_path): + print("Error: Copy Failure: destination path does not exist: ", destination_path) + raise ValidationError(f"Error: Copy Failure: destination path does not exist: {destination_path}") + + try: + if verbose: + print(f"Copying: {source_file} to {destination_path}") + if path.isdir(source_file): + copy_tree(source_file, destination_path) + else: + shutil.copy2(source_file, destination_path) + except IOError as e: + msg = f"Error: copy failure.\n copying: {source_file}\n to: {destination_path}" + raise IOError(msg) from e + return 0 + + +def copy_multiple_items_to_multiple_dirs(source_paths: [str], destination_paths: [str], verbose: bool=False): + """ Copies the sources to the destination paths. + +This is a generic function that just wraps shutil.copy2""" + + if source_paths is None: + raise ValidationError("Error: source for copy cannot be None type") + if destination_paths is None: + raise ValidationError("Error: copy destination cannot be None type") + + if type(source_paths) is not list or len(source_paths) == 0: + raise ValidationError("Error: sources must be given as a list with at least one entry") + if type(destination_paths) is not list or len(destination_paths) == 0: + raise ValidationError("Error: destinations must be given as a list with at least one entry") + + for dpath in destination_paths: + if not path.exists(dpath): + if verbose: + print("Copy Warning: target path does not exist (skipping): ", dpath) + continue + if not path.isdir(dpath): + if verbose: + print("Copy Warning: target path is not a directory (skipping): ", ) + continue + + for spath in source_paths: + result = copy_to_dir(spath, dpath, verbose) + if result != 0: + return result + + return 0 Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/generic_utils_test.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/generic_utils_test.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/generic_utils_test.py (revision 40898) @@ -0,0 +1,224 @@ +import backups.utils.generic_utils as utils +import os +import pytest + +from backups.utils.errors import ValidationError +from pathlib import Path + + +def make_a_real_file(file_path, file_name): + real_file = os.path.join(file_path, file_name) + Path(real_file).touch() + assert os.path.exists(real_file) + assert os.path.isfile(real_file) + return real_file + + +class TestCopyingToDir(object): + + def test_copy_fails_with_none_args(self, tmpdir): + with pytest.raises(ValidationError): + utils.copy_to_dir(None, tmpdir) + + with pytest.raises(ValidationError): + real_file = make_a_real_file(tmpdir, "file.txt") + utils.copy_to_dir(real_file, None) + + def test_copy_fails_if_source_does_not_exist(self, tmpdir): + with pytest.raises(ValidationError): + utils.copy_to_dir("not a real file", tmpdir) + + def test_copy_fails_if_destination_is_not_real(self, tmpdir): + with pytest.raises(ValidationError): + real_file = make_a_real_file(tmpdir, "real_file.txt") + utils.copy_to_dir(real_file, "not a real directory") + + def test_copy_fails_if_destination_is_not_a_directory(self, tmpdir): + + real_file = make_a_real_file(tmpdir, "not_a_directory.txt") + not_a_directory = real_file + assert os.path.exists(not_a_directory) + assert not os.path.isdir(not_a_directory) + + with pytest.raises(ValidationError): + utils.copy_to_dir(real_file, not_a_directory) + + def test_copy_to_dir(self, tmpdir): + + # Setup source file and target dir + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + assert os.path.exists(destination_dir) + file_name = "fun_file.txt" + file_to_copy = os.path.join(tmpdir, file_name) + Path(file_to_copy).touch() + assert os.path.exists(file_to_copy) + + # copied file does not exist yet + expected_copy = os.path.join(tmpdir, "destination", file_name) + assert not os.path.exists(expected_copy) + + # run copy function and confirm copy exists (and original) + result = utils.copy_to_dir(file_to_copy, destination_dir) + assert result == 0 + assert os.path.exists(file_to_copy) + assert os.path.exists(expected_copy) + + def test_copying_directories_and_subdirectories(self, tmpdir): + + subdir = os.path.join(tmpdir, "subdir") + subsubdir = os.path.join(subdir, "subsubdir") + os.makedirs(subsubdir) + filename_a = "file_in_subdir.txt" + filename_b = "file_in_subsubdir.txt" + filepath_a = make_a_real_file(subdir, filename_a) + filepath_b = make_a_real_file(subsubdir, filename_b) + + print(filepath_a) + print(filepath_b) + + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + + expected_filepath_a = os.path.join(destination_dir, "subdir", filename_a) + expected_filepath_b = os.path.join(destination_dir, "subdir", "subsubdir", filename_b) + print(expected_filepath_a) + print(expected_filepath_b) + print(os.path.exists(expected_filepath_a)) + if os.path.exists(expected_filepath_a): + print("apple") + else: + print("banana") + print(os.path.exists(expected_filepath_b)) + + os.path.commonpath + + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + + # Provide only the original tmpdir to the copying function + utils.copy_to_dir(subdir, destination_dir, True) + + # assert that the subdirs and files all exist + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + + +class TestMultiCopy(object): + + def test_copy_fails_if_args_are_none(self, tmpdir): + + with pytest.raises(ValidationError): + utils.copy_multiple_items_to_multiple_dirs(None, None) + + with pytest.raises(ValidationError): + utils.copy_multiple_items_to_multiple_dirs(tmpdir, None) + + with pytest.raises(ValidationError): + utils.copy_multiple_items_to_multiple_dirs(None, tmpdir) + + def test_copy_requires_args_to_be_in_list_format(self, tmpdir): + + not_a_list = "not a list" + with pytest.raises(ValidationError): + utils.copy_multiple_items_to_multiple_dirs(not_a_list, tmpdir) + + tmpfile = make_a_real_file(tmpdir, "a_file.txt") + with pytest.raises(ValidationError): + utils.copy_multiple_items_to_multiple_dirs(tmpfile, not_a_list) + + def test_copy_is_fine_with_single_items(self, tmpdir): + + tmpfile = make_a_real_file(tmpdir, "single_file.txt") + destination = os.path.join(tmpdir, "destination") + os.makedirs(destination) + expected_file = os.path.join(tmpdir, "destination", "single_file.txt") + assert not os.path.exists(expected_file) + + result = utils.copy_multiple_items_to_multiple_dirs([tmpfile], [destination]) + assert result == 0 + assert os.path.exists(expected_file) + + def test_copy_multiple_items_to_multiple_dirs(self, tmpdir): + + file_a = make_a_real_file(tmpdir, "single_file.txt") + file_b = make_a_real_file(tmpdir, "single_file.txt") + file_c = make_a_real_file(tmpdir, "single_file.txt") + source_files = [ file_a + , file_b + , file_c + ] + + destination_1 = os.path.join(tmpdir, "dest_1") + destination_2 = os.path.join(tmpdir, "dest_2") + destination_3 = os.path.join(tmpdir, "dest_3") + dest_paths = [ destination_1 + , destination_2 + , destination_3 + ] + os.makedirs(destination_1) + os.makedirs(destination_2) + os.makedirs(destination_3) + + expected_files = [] + for dp in dest_paths: + for sf in source_files: + expected_files.append( os.path.join(dp, os.path.basename(sf))) + assert not os.path.exists(expected_files[-1]) + assert len(expected_files) == 9 + + result = utils.copy_multiple_items_to_multiple_dirs(source_files, dest_paths) + assert result == 0 + for ef in expected_files: + assert os.path.exists(ef) + + def test_copying_is_recursive(self, tmpdir): + + # make some subdirs, with sub-subdirs, and files within them + subdir_a = os.path.join(tmpdir, "subdir_a") + subsubdir_aa = os.path.join(subdir_a, "subsubdir_aa") + subsubdir_ab = os.path.join(subdir_a, "subsubdir_ab") + subdir_b = os.path.join(tmpdir, "subdir_b") + subsubdir_ba = os.path.join(subdir_b, "subsubdir_ba") + subsubdir_bb = os.path.join(subdir_b, "subsubdir_bb") + os.makedirs(subsubdir_aa) + os.makedirs(subsubdir_ab) + os.makedirs(subsubdir_ba) + os.makedirs(subsubdir_bb) + make_a_real_file(subdir_a, "file_a.txt") + make_a_real_file(subsubdir_aa, "file_aa.txt") + make_a_real_file(subsubdir_ab, "file_ab.txt") + make_a_real_file(subdir_b, "file_b.txt") + make_a_real_file(subsubdir_ba, "file_ba.txt") + make_a_real_file(subsubdir_bb, "file_bb.txt") + + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + + expected_filepath_a = os.path.join(destination_dir, "subdir_a", "file_a.txt") + expected_filepath_aa = os.path.join(destination_dir, "subdir_a", "subsubdir_aa", "file_aa.txt") + expected_filepath_ab = os.path.join(destination_dir, "subdir_a", "subsubdir_ab", "file_ab.txt") + expected_filepath_b = os.path.join(destination_dir, "subdir_b", "file_b.txt") + expected_filepath_ba = os.path.join(destination_dir, "subdir_b", "subsubdir_ba", "file_ba.txt") + expected_filepath_bb = os.path.join(destination_dir, "subdir_b", "subsubdir_bb", "file_bb.txt") + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_aa) + assert not os.path.exists(expected_filepath_ab) + assert not os.path.exists(expected_filepath_b) + assert not os.path.exists(expected_filepath_ba) + assert not os.path.exists(expected_filepath_bb) + + # Provide only the original tmpdir to the copying function + utils.copy_multiple_items_to_multiple_dirs([tmpdir], [destination_dir], True) + + # assert that the subdirs and files all exist + assert os.path.exists(expected_filepath_a) + assert os.path.exists(expected_filepath_b) + assert os.path.exists(expected_filepath_a) + assert os.path.exists(expected_filepath_aa) + assert os.path.exists(expected_filepath_ab) + assert os.path.exists(expected_filepath_b) + assert os.path.exists(expected_filepath_ba) + assert os.path.exists(expected_filepath_bb) Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/generic_utils.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/generic_utils.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/generic_utils.py (revision 40898) @@ -0,0 +1,70 @@ +import os.path as path +import shutil + +from distutils.dir_util import copy_tree + + +class ValidationError(Exception): + """validation has failed in the generic utils module""" + + +def copy_to_dir(source_file: str, destination_path: str, verbose: bool=False): + """ Copies a single file into the destination directory. If a directory is + given as a source then all the contents of the directory will be copied (the + directory itself will not be). +""" + + if source_file is None or destination_path is None: + raise ValidationError("Error: source for copy cannot be None type") + if destination_path is None: + raise ValidationError("Error: copy destination cannot be None type") + if not path.exists(source_file): + raise ValidationError(f"Error: Copy Failure: source does not exist: {source_file}") + if not path.exists(destination_path) or not path.isdir(destination_path): + print("Error: Copy Failure: destination path does not exist: ", destination_path) + raise ValidationError(f"Error: Copy Failure: destination path does not exist: {destination_path}") + + try: + if verbose: + print(f"Copying: {source_file} to {destination_path}") + if path.isdir(source_file): + copy_tree(source_file, destination_path) + else: + shutil.copy2(source_file, destination_path) + except IOError as e: + msg = f"Error: copy failure.\n copying: {source_file}\n to: {destination_path}" + raise IOError(msg) from e + return 0 + + +def copy_multiple_items_to_multiple_dirs(source_paths: [str], destination_paths: [str], verbose: bool=False): + """ Copies the sources to the destination paths. + +This is a generic function that just wraps shutil.copy2""" + + if source_paths is None: + raise ValidationError("Error: source for copy cannot be None type") + if destination_paths is None: + raise ValidationError("Error: copy destination cannot be None type") + + if type(source_paths) is not list or len(source_paths) == 0: + raise ValidationError("Error: sources must be given as a list with at least one entry") + if type(destination_paths) is not list or len(destination_paths) == 0: + raise ValidationError("Error: destinations must be given as a list with at least one entry") + + for dpath in destination_paths: + if not path.exists(dpath): + if verbose: + print("Copy Warning: target path does not exist (skipping): ", dpath) + continue + if not path.isdir(dpath): + if verbose: + print("Copy Warning: target path is not a directory (skipping): ", ) + continue + + for spath in source_paths: + result = copy_to_dir(spath, dpath, verbose) + if result != 0: + return result + + return 0 Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/generic_utils_test.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/generic_utils_test.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/generic_utils_test.py (revision 40898) @@ -0,0 +1,224 @@ +import generic_utils as utils +import os +import pytest +import subprocess + +from pathlib import Path + + +def make_a_real_file(file_path, file_name): + real_file = os.path.join(file_path, file_name) + Path(real_file).touch() + assert os.path.exists(real_file) + assert os.path.isfile(real_file) + return real_file + + +class TestCopyingToDir(object): + + def test_copy_fails_with_none_args(self, tmpdir): + with pytest.raises(utils.ValidationError): + utils.copy_to_dir(None, tmpdir) + + with pytest.raises(utils.ValidationError): + real_file = make_a_real_file(tmpdir, "file.txt") + utils.copy_to_dir(real_file, None) + + def test_copy_fails_if_source_does_not_exist(self, tmpdir): + with pytest.raises(utils.ValidationError): + utils.copy_to_dir("not a real file", tmpdir) + + def test_copy_fails_if_destination_is_not_real(self, tmpdir): + with pytest.raises(utils.ValidationError): + real_file = make_a_real_file(tmpdir, "real_file.txt") + utils.copy_to_dir(real_file, "not a real directory") + + def test_copy_fails_if_destination_is_not_a_directory(self, tmpdir): + + real_file = make_a_real_file(tmpdir, "not_a_directory.txt") + not_a_directory = real_file + assert os.path.exists(not_a_directory) + assert not os.path.isdir(not_a_directory) + + with pytest.raises(utils.ValidationError): + utils.copy_to_dir(real_file, not_a_directory) + + def test_copy_to_dir(self, tmpdir): + + # Setup source file and target dir + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + assert os.path.exists(destination_dir) + file_name = "fun_file.txt" + file_to_copy = os.path.join(tmpdir, file_name) + Path(file_to_copy).touch() + assert os.path.exists(file_to_copy) + + # copied file does not exist yet + expected_copy = os.path.join(tmpdir, "destination", file_name) + assert not os.path.exists(expected_copy) + + # run copy function and confirm copy exists (and original) + result = utils.copy_to_dir(file_to_copy, destination_dir) + assert result == 0 + assert os.path.exists(file_to_copy) + assert os.path.exists(expected_copy) + + def test_copying_directories_and_subdirectories(self, tmpdir): + + subdir = os.path.join(tmpdir, "subdir") + subsubdir = os.path.join(subdir, "subsubdir") + os.makedirs(subsubdir) + filename_a = "file_in_subdir.txt" + filename_b = "file_in_subsubdir.txt" + filepath_a = make_a_real_file(subdir, filename_a) + filepath_b = make_a_real_file(subsubdir, filename_b) + + print(filepath_a) + print(filepath_b) + + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + + expected_filepath_a = os.path.join(destination_dir, "subdir", filename_a) + expected_filepath_b = os.path.join(destination_dir, "subdir", "subsubdir", filename_b) + print(expected_filepath_a) + print(expected_filepath_b) + print(os.path.exists(expected_filepath_a)) + if os.path.exists(expected_filepath_a): + print("apple") + else: + print("banana") + print(os.path.exists(expected_filepath_b)) + + os.path.commonpath + + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + + # Provide only the original tmpdir to the copying function + utils.copy_to_dir(subdir, destination_dir, True) + + # assert that the subdirs and files all exist + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + + +class TestMultiCopy(object): + + def test_copy_fails_if_args_are_none(self, tmpdir): + + with pytest.raises(utils.ValidationError): + utils.copy_multiple_items_to_multiple_dirs(None, None) + + with pytest.raises(utils.ValidationError): + utils.copy_multiple_items_to_multiple_dirs(tmpdir, None) + + with pytest.raises(utils.ValidationError): + utils.copy_multiple_items_to_multiple_dirs(None, tmpdir) + + def test_copy_requires_args_to_be_in_list_format(self, tmpdir): + + not_a_list = "not a list" + with pytest.raises(utils.ValidationError): + utils.copy_multiple_items_to_multiple_dirs(not_a_list, tmpdir) + + tmpfile = make_a_real_file(tmpdir, "a_file.txt") + with pytest.raises(utils.ValidationError): + utils.copy_multiple_items_to_multiple_dirs(tmpfile, not_a_list) + + def test_copy_is_fine_with_single_items(self, tmpdir): + + tmpfile = make_a_real_file(tmpdir, "single_file.txt") + destination = os.path.join(tmpdir, "destination") + os.makedirs(destination) + expected_file = os.path.join(tmpdir, "destination", "single_file.txt") + assert not os.path.exists(expected_file) + + result = utils.copy_multiple_items_to_multiple_dirs([tmpfile], [destination]) + assert result == 0 + assert os.path.exists(expected_file) + + def test_copy_multiple_items_to_multiple_dirs(self, tmpdir): + + file_a = make_a_real_file(tmpdir, "single_file.txt") + file_b = make_a_real_file(tmpdir, "single_file.txt") + file_c = make_a_real_file(tmpdir, "single_file.txt") + source_files = [ file_a + , file_b + , file_c + ] + + destination_1 = os.path.join(tmpdir, "dest_1") + destination_2 = os.path.join(tmpdir, "dest_2") + destination_3 = os.path.join(tmpdir, "dest_3") + dest_paths = [ destination_1 + , destination_2 + , destination_3 + ] + os.makedirs(destination_1) + os.makedirs(destination_2) + os.makedirs(destination_3) + + expected_files = [] + for dp in dest_paths: + for sf in source_files: + expected_files.append( os.path.join(dp, os.path.basename(sf))) + assert not os.path.exists(expected_files[-1]) + assert len(expected_files) == 9 + + result = utils.copy_multiple_items_to_multiple_dirs(source_files, dest_paths) + assert result == 0 + for ef in expected_files: + assert os.path.exists(ef) + + def test_copying_is_recursive(self, tmpdir): + + # make some subdirs, with sub-subdirs, and files within them + subdir_a = os.path.join(tmpdir, "subdir_a") + subsubdir_aa = os.path.join(subdir_a, "subsubdir_aa") + subsubdir_ab = os.path.join(subdir_a, "subsubdir_ab") + subdir_b = os.path.join(tmpdir, "subdir_b") + subsubdir_ba = os.path.join(subdir_b, "subsubdir_ba") + subsubdir_bb = os.path.join(subdir_b, "subsubdir_bb") + os.makedirs(subsubdir_aa) + os.makedirs(subsubdir_ab) + os.makedirs(subsubdir_ba) + os.makedirs(subsubdir_bb) + make_a_real_file(subdir_a, "file_a.txt") + make_a_real_file(subsubdir_aa, "file_aa.txt") + make_a_real_file(subsubdir_ab, "file_ab.txt") + make_a_real_file(subdir_b, "file_b.txt") + make_a_real_file(subsubdir_ba, "file_ba.txt") + make_a_real_file(subsubdir_bb, "file_bb.txt") + + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + + expected_filepath_a = os.path.join(destination_dir, "subdir_a", "file_a.txt") + expected_filepath_aa = os.path.join(destination_dir, "subdir_a", "subsubdir_aa", "file_aa.txt") + expected_filepath_ab = os.path.join(destination_dir, "subdir_a", "subsubdir_ab", "file_ab.txt") + expected_filepath_b = os.path.join(destination_dir, "subdir_b", "file_b.txt") + expected_filepath_ba = os.path.join(destination_dir, "subdir_b", "subsubdir_ba", "file_ba.txt") + expected_filepath_bb = os.path.join(destination_dir, "subdir_b", "subsubdir_bb", "file_bb.txt") + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_aa) + assert not os.path.exists(expected_filepath_ab) + assert not os.path.exists(expected_filepath_b) + assert not os.path.exists(expected_filepath_ba) + assert not os.path.exists(expected_filepath_bb) + + # Provide only the original tmpdir to the copying function + utils.copy_multiple_items_to_multiple_dirs([tmpdir], [destination_dir], True) + + # assert that the subdirs and files all exist + assert os.path.exists(expected_filepath_a) + assert os.path.exists(expected_filepath_b) + assert os.path.exists(expected_filepath_a) + assert os.path.exists(expected_filepath_aa) + assert os.path.exists(expected_filepath_ab) + assert os.path.exists(expected_filepath_b) + assert os.path.exists(expected_filepath_ba) + assert os.path.exists(expected_filepath_bb) Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/subprocess_utils.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/subprocess_utils.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/subprocess_utils.py (revision 40898) @@ -0,0 +1,53 @@ +import subprocess + +# Plans are to flesh this out a bit more to make some smarter copying etc +from generic_utils import ValidationError + + +class SubprocessError(Exception): + """Errors in Popen stderr subprocess""" + + +def _args_check(args, function_name): + if args is None: + raise ValidationError(f"Error: no arguments provided in {function_name}") + if type(args) != list or len(args) == 0: + raise ValidationError(f"Error: no arguments provided in {function_name}") + return 0 + + +def cp_wrapper(args: [str]): + if args is None: + raise ValidationError(f"Error: no arguments provided to {cp_wrapper.__name__}") + command_and_args = ["cp"] + args + return simple_unix_wrapper(command_and_args) + + +def tar_wrapper(args: [str]): + if args is None: + raise ValidationError(f"Error: no arguments provided to {tar_wrapper.__name__}") + command_and_args = ["tar"] + args + return simple_unix_wrapper(command_and_args) + + +def chmod_wrapper(args: [str]): + if args is None: + raise ValidationError(f"Error: no arguments provided to {chmod_wrapper.__name__}") + command_and_args = ["chmod"] + args + return simple_unix_wrapper(command_and_args) + + +def simple_unix_wrapper(command_and_args: [str]): + """ This should provide the program name and all arguments in a list. + +Only a single command should be called. + """ + _args_check(command_and_args, simple_unix_wrapper.__name__) + + try: + subprocess.check_call(command_and_args) + except (subprocess.CalledProcessError) as cpe: + raise SubprocessError(f"{type(cpe)} in python simple_unix_wrapper") from cpe + except (ValidationError) as ve: + raise ValidationError(f"{type(ve)} in python simple_unix_wrapper") from ve + return 0 Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/subprocess_utils_test.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/subprocess_utils_test.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/python/subprocess_utils_test.py (revision 40898) @@ -0,0 +1,165 @@ +import subprocess_utils as utils +import os +import pytest +import stat + +from pathlib import Path +from generic_utils import ValidationError + + +def make_a_real_file(file_path, file_name): + real_file = os.path.join(file_path, file_name) + Path(real_file).touch() + assert os.path.exists(real_file) + assert os.path.isfile(real_file) + return real_file + + +class TestCopyWrapper(object): + + def test_copy_raises_validation_error_with_no_args(self): + with pytest.raises(ValidationError): + utils.cp_wrapper(None) + + def test_copy_fails_with_invalid_args(self, tmpdir): + with pytest.raises(utils.SubprocessError): + utils.cp_wrapper(["not a real file", "not a valid copy_location"]) + + def test_copy_fails_if_source_does_not_exist(self, tmpdir): + with pytest.raises(utils.SubprocessError): + utils.cp_wrapper(["not a real file", tmpdir]) + + def test_simple_copy(self, tmpdir): + real_file = make_a_real_file(tmpdir, "real_file.txt") + expected_filepath = os.path.join(tmpdir, "copied_file.txt") + assert not os.path.isfile(expected_filepath) + + result = utils.cp_wrapper(["-v", real_file, expected_filepath]) + assert os.path.isfile(expected_filepath) + assert result == 0 + + def test_simple_copy_to_directory(self, tmpdir): + + # Setup source file and target dir + filename = "real_file.txt" + real_file = make_a_real_file(tmpdir, filename) + + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + assert os.path.exists(destination_dir) + + expected_filepath = os.path.join(destination_dir, filename) + assert not os.path.exists(expected_filepath) + + # run copy function and confirm copy exists (and original) + result = utils.cp_wrapper([real_file, destination_dir]) + assert result == 0 + assert os.path.isfile(real_file) + assert os.path.exists(expected_filepath) + + def test_copying_directories_and_subdirectories(self, tmpdir): + + subdir = os.path.join(tmpdir, "subdir") + subsubdir = os.path.join(subdir, "subsubdir") + os.makedirs(subsubdir) + filename_a = "file_in_subdir.txt" + filename_b = "file_in_subsubdir.txt" + make_a_real_file(subdir, filename_a) + make_a_real_file(subsubdir, filename_b) + + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + expected_filepath_a = os.path.join(destination_dir, "subdir", filename_a) + expected_filepath_b = os.path.join(destination_dir, "subdir", "subsubdir", filename_b) + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + + # fails without -r arg: + with pytest.raises(utils.SubprocessError): + utils.cp_wrapper([subdir, destination_dir]) + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + + utils.cp_wrapper(["-r", subdir, destination_dir]) + assert os.path.exists(expected_filepath_a) + assert os.path.exists(expected_filepath_b) + + +class TestTarWrapper(object): + + def test_tar_raise_validation_error_with_none_args(self): + with pytest.raises(ValidationError): + utils.tar_wrapper(None) + + def test_tar_returns_safely_with_too_few_args(self): + with pytest.raises(utils.SubprocessError): + utils.tar_wrapper(["only one arg"]) + + def test_tar_return_error_if_invalid_args_given(self, tmpdir): + tar_filename = os.path.join(tmpdir, "tar_file.tar") + fake_file = os.path.join(tmpdir, "fake_file.txt") + assert not os.path.exists(fake_file) + with pytest.raises(utils.SubprocessError): + utils.tar_wrapper(["-cf", tar_filename, fake_file]) + + def test_simple_tar_of_files(self, tmpdir): + file_a = make_a_real_file(tmpdir, "file_a.txt") + file_b = make_a_real_file(tmpdir, "file_b.txt") + + expected_file = os.path.join(tmpdir, "output.tar") + assert not os.path.exists(expected_file) + + args = ["-cf", expected_file, file_a, file_b] + utils.tar_wrapper(args) + assert os.path.exists(expected_file) + + +class TestChmodWrapper(object): + + def test_chmod_gives_validation_error_when_no_args(self): + with pytest.raises(utils.ValidationError): + utils.chmod_wrapper(None) + + def test_chmod_wrapper_raises_error_with_invalid_args(self): + with pytest.raises(utils.SubprocessError): + utils.chmod_wrapper(["only one arg"]) + + def test_chmod_fails_if_no_permissions_given(self, tmpdir): + real_file = make_a_real_file(tmpdir, 'mr_file') + + with pytest.raises(utils.SubprocessError): + utils.chmod_wrapper([real_file]) + + with pytest.raises(utils.SubprocessError): + utils.chmod_wrapper(["dumb_arg", real_file]) + + def test_can_chmod_file(self, tmpdir): + real_file = make_a_real_file(tmpdir, 'mr_file') + + # check initial permissions + org_stats = os.stat(real_file) + # Default should be: rw-rw-r-- + assert bool(org_stats.st_mode & stat.S_IRUSR) is True + assert bool(org_stats.st_mode & stat.S_IWUSR) is True + assert bool(org_stats.st_mode & stat.S_IXUSR) is False + assert bool(org_stats.st_mode & stat.S_IRGRP) is True + assert bool(org_stats.st_mode & stat.S_IWGRP) is True + assert bool(org_stats.st_mode & stat.S_IXGRP) is False + assert bool(org_stats.st_mode & stat.S_IROTH) is True + assert bool(org_stats.st_mode & stat.S_IWOTH) is False + assert bool(org_stats.st_mode & stat.S_IXOTH) is False + + # change them + utils.chmod_wrapper(["777", real_file]) + + # check they changed + new_stats = os.stat(real_file) + assert bool(new_stats.st_mode & stat.S_IRUSR) is True + assert bool(new_stats.st_mode & stat.S_IWUSR) is True + assert bool(new_stats.st_mode & stat.S_IXUSR) is True + assert bool(new_stats.st_mode & stat.S_IRGRP) is True + assert bool(new_stats.st_mode & stat.S_IWGRP) is True + assert bool(new_stats.st_mode & stat.S_IXGRP) is True + assert bool(new_stats.st_mode & stat.S_IROTH) is True + assert bool(new_stats.st_mode & stat.S_IWOTH) is True + assert bool(new_stats.st_mode & stat.S_IXOTH) is True Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/subprocess_utils.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/subprocess_utils.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/subprocess_utils.py (revision 40898) @@ -0,0 +1,49 @@ +import subprocess + +# Plans are to flesh this out a bit more to make some smarter copying etc +from backups.utils.errors import ValidationError, SubprocessError + + +def _args_check(args, function_name): + if args is None: + raise ValidationError(f"Error: no arguments provided in {function_name}") + if type(args) != list or len(args) == 0: + raise ValidationError(f"Error: no arguments provided in {function_name}") + return 0 + + +def cp_wrapper(args: [str]): + if args is None: + raise ValidationError(f"Error: no arguments provided to {cp_wrapper.__name__}") + command_and_args = ["cp"] + args + return simple_unix_wrapper(command_and_args) + + +def tar_wrapper(args: [str]): + if args is None: + raise ValidationError(f"Error: no arguments provided to {tar_wrapper.__name__}") + command_and_args = ["tar"] + args + return simple_unix_wrapper(command_and_args) + + +def chmod_wrapper(args: [str]): + if args is None: + raise ValidationError(f"Error: no arguments provided to {chmod_wrapper.__name__}") + command_and_args = ["chmod"] + args + return simple_unix_wrapper(command_and_args) + + +def simple_unix_wrapper(command_and_args: [str]): + """ This should provide the program name and all arguments in a list. + +Only a single command should be called. + """ + _args_check(command_and_args, simple_unix_wrapper.__name__) + + try: + subprocess.check_call(command_and_args) + except (subprocess.CalledProcessError) as cpe: + raise SubprocessError(f"{type(cpe)} in python simple_unix_wrapper") from cpe + except (ValidationError) as ve: + raise ValidationError(f"{type(ve)} in python simple_unix_wrapper") from ve + return 0 Index: /branches/ipp-132_automate_jira_conf_backups/backups/utils/subprocess_utils_test.py =================================================================== --- /branches/ipp-132_automate_jira_conf_backups/backups/utils/subprocess_utils_test.py (revision 40898) +++ /branches/ipp-132_automate_jira_conf_backups/backups/utils/subprocess_utils_test.py (revision 40898) @@ -0,0 +1,165 @@ +import backups.utils.subprocess_utils as utils +import os +import pytest +import stat + +from pathlib import Path +from backups.utils.errors import SubprocessError, ValidationError + + +def make_a_real_file(file_path, file_name): + real_file = os.path.join(file_path, file_name) + Path(real_file).touch() + assert os.path.exists(real_file) + assert os.path.isfile(real_file) + return real_file + + +class TestCopyWrapper(object): + + def test_copy_raises_validation_error_with_no_args(self): + with pytest.raises(ValidationError): + utils.cp_wrapper(None) + + def test_copy_fails_with_invalid_args(self, tmpdir): + with pytest.raises(SubprocessError): + utils.cp_wrapper(["not a real file", "not a valid copy_location"]) + + def test_copy_fails_if_source_does_not_exist(self, tmpdir): + with pytest.raises(SubprocessError): + utils.cp_wrapper(["not a real file", tmpdir]) + + def test_simple_copy(self, tmpdir): + real_file = make_a_real_file(tmpdir, "real_file.txt") + expected_filepath = os.path.join(tmpdir, "copied_file.txt") + assert not os.path.isfile(expected_filepath) + + result = utils.cp_wrapper(["-v", real_file, expected_filepath]) + assert os.path.isfile(expected_filepath) + assert result == 0 + + def test_simple_copy_to_directory(self, tmpdir): + + # Setup source file and target dir + filename = "real_file.txt" + real_file = make_a_real_file(tmpdir, filename) + + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + assert os.path.exists(destination_dir) + + expected_filepath = os.path.join(destination_dir, filename) + assert not os.path.exists(expected_filepath) + + # run copy function and confirm copy exists (and original) + result = utils.cp_wrapper([real_file, destination_dir]) + assert result == 0 + assert os.path.isfile(real_file) + assert os.path.exists(expected_filepath) + + def test_copying_directories_and_subdirectories(self, tmpdir): + + subdir = os.path.join(tmpdir, "subdir") + subsubdir = os.path.join(subdir, "subsubdir") + os.makedirs(subsubdir) + filename_a = "file_in_subdir.txt" + filename_b = "file_in_subsubdir.txt" + make_a_real_file(subdir, filename_a) + make_a_real_file(subsubdir, filename_b) + + destination_dir = os.path.join(tmpdir, "destination") + os.makedirs(destination_dir) + expected_filepath_a = os.path.join(destination_dir, "subdir", filename_a) + expected_filepath_b = os.path.join(destination_dir, "subdir", "subsubdir", filename_b) + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + + # fails without -r arg: + with pytest.raises(SubprocessError): + utils.cp_wrapper([subdir, destination_dir]) + assert not os.path.exists(expected_filepath_a) + assert not os.path.exists(expected_filepath_b) + + utils.cp_wrapper(["-r", subdir, destination_dir]) + assert os.path.exists(expected_filepath_a) + assert os.path.exists(expected_filepath_b) + + +class TestTarWrapper(object): + + def test_tar_raise_validation_error_with_none_args(self): + with pytest.raises(ValidationError): + utils.tar_wrapper(None) + + def test_tar_returns_safely_with_too_few_args(self): + with pytest.raises(SubprocessError): + utils.tar_wrapper(["only one arg"]) + + def test_tar_return_error_if_invalid_args_given(self, tmpdir): + tar_filename = os.path.join(tmpdir, "tar_file.tar") + fake_file = os.path.join(tmpdir, "fake_file.txt") + assert not os.path.exists(fake_file) + with pytest.raises(SubprocessError): + utils.tar_wrapper(["-cf", tar_filename, fake_file]) + + def test_simple_tar_of_files(self, tmpdir): + file_a = make_a_real_file(tmpdir, "file_a.txt") + file_b = make_a_real_file(tmpdir, "file_b.txt") + + expected_file = os.path.join(tmpdir, "output.tar") + assert not os.path.exists(expected_file) + + args = ["-cf", expected_file, file_a, file_b] + utils.tar_wrapper(args) + assert os.path.exists(expected_file) + + +class TestChmodWrapper(object): + + def test_chmod_gives_validation_error_when_no_args(self): + with pytest.raises(utils.ValidationError): + utils.chmod_wrapper(None) + + def test_chmod_wrapper_raises_error_with_invalid_args(self): + with pytest.raises(SubprocessError): + utils.chmod_wrapper(["only one arg"]) + + def test_chmod_fails_if_no_permissions_given(self, tmpdir): + real_file = make_a_real_file(tmpdir, 'mr_file') + + with pytest.raises(SubprocessError): + utils.chmod_wrapper([real_file]) + + with pytest.raises(SubprocessError): + utils.chmod_wrapper(["dumb_arg", real_file]) + + def test_can_chmod_file(self, tmpdir): + real_file = make_a_real_file(tmpdir, 'mr_file') + + # check initial permissions + org_stats = os.stat(real_file) + # Default should be: rw-rw-r-- + assert bool(org_stats.st_mode & stat.S_IRUSR) is True + assert bool(org_stats.st_mode & stat.S_IWUSR) is True + assert bool(org_stats.st_mode & stat.S_IXUSR) is False + assert bool(org_stats.st_mode & stat.S_IRGRP) is True + assert bool(org_stats.st_mode & stat.S_IWGRP) is True + assert bool(org_stats.st_mode & stat.S_IXGRP) is False + assert bool(org_stats.st_mode & stat.S_IROTH) is True + assert bool(org_stats.st_mode & stat.S_IWOTH) is False + assert bool(org_stats.st_mode & stat.S_IXOTH) is False + + # change them + utils.chmod_wrapper(["777", real_file]) + + # check they changed + new_stats = os.stat(real_file) + assert bool(new_stats.st_mode & stat.S_IRUSR) is True + assert bool(new_stats.st_mode & stat.S_IWUSR) is True + assert bool(new_stats.st_mode & stat.S_IXUSR) is True + assert bool(new_stats.st_mode & stat.S_IRGRP) is True + assert bool(new_stats.st_mode & stat.S_IWGRP) is True + assert bool(new_stats.st_mode & stat.S_IXGRP) is True + assert bool(new_stats.st_mode & stat.S_IROTH) is True + assert bool(new_stats.st_mode & stat.S_IWOTH) is True + assert bool(new_stats.st_mode & stat.S_IXOTH) is True