diff --git a/hed/tools/remodeling/backup_manager.py b/hed/tools/remodeling/backup_manager.py index 608f4331b..d98b77c0d 100644 --- a/hed/tools/remodeling/backup_manager.py +++ b/hed/tools/remodeling/backup_manager.py @@ -10,16 +10,16 @@ class BackupManager: DEFAULT_BACKUP_NAME = 'default_back' - RELATIVE_BACKUP_LOCATION = 'derivatives/remodel/backups' + RELATIVE_BACKUP_LOCATION = 'derivatives/remodel' BACKUP_DICTIONARY = 'backup_lock.json' BACKUP_ROOT = 'backup_root' - def __init__(self, data_root): + def __init__(self, data_root, backups_root=None): """ Constructor for the backup manager. Parameters: - data_root (str): full path of the root of the data directory. - + data_root (str): Full path of the root of the data directory. + backups_root (str or None): Full path to the root where backups subdirectory is located. Raises: - HedFileError: - If the data_root does not correspond to a real directory. @@ -28,11 +28,15 @@ def __init__(self, data_root): if not os.path.isdir(data_root): raise HedFileError('NonExistentData', f"{data_root} is not an existing directory", "") self.data_root = data_root - self.backups_root = os.path.join(data_root, self.RELATIVE_BACKUP_LOCATION) - os.makedirs(self.backups_root, exist_ok=True) + if backups_root: + self.backups_path = os.path.join(backups_root, 'backups') + else: + self.backups_path = os.path.join(data_root, self.RELATIVE_BACKUP_LOCATION, 'backups') + self.backups_path = os.path.realpath(self.backups_path) + os.makedirs(self.backups_path, exist_ok=True) self.backups_dict = self._get_backups() - def create_backup(self, file_list, backup_name=None, verbose=True): + def create_backup(self, file_list, backup_name=None, verbose=False): """ Create a new backup from file_list. Parameters: @@ -46,7 +50,7 @@ def create_backup(self, file_list, backup_name=None, verbose=True): Raises: HedFileError - For missing or incorrect files. - + OS-related error - OS-related error when file copying occurs. @@ -59,7 +63,7 @@ def create_backup(self, file_list, backup_name=None, verbose=True): time_stamp = f"{str(datetime.now())}" if verbose: print(f"Creating backup {backup_name}") - backup_dir_path = os.path.realpath(os.path.join(self.backups_root, backup_name, BackupManager.BACKUP_ROOT)) + backup_dir_path = os.path.realpath(os.path.join(self.backups_path, backup_name, BackupManager.BACKUP_ROOT)) os.makedirs(backup_dir_path, exist_ok=True) for file in file_list: backup_file = self.get_backup_path(backup_name, file) @@ -69,7 +73,7 @@ def create_backup(self, file_list, backup_name=None, verbose=True): shutil.copy2(file, backup_file) backup[self.get_file_key(file)] = time_stamp self.backups_dict[backup_name] = backup - backup_dict_path = os.path.realpath(os.path.join(self.backups_root, backup_name, + backup_dict_path = os.path.realpath(os.path.join(self.backups_path, backup_name, self.BACKUP_DICTIONARY)) with open(backup_dict_path, 'w') as fp: json.dump(backup, fp, indent=4) @@ -83,11 +87,11 @@ def get_backup(self, backup_name): Returns: The dictionary with the backup info. - + Notes: - - The dictionary with backup information has keys that are the paths of - the backed up files relative to the backup root. The values in this - dictionary are the dates on which the particular file was backed up. + The dictionary with backup information has keys that are the paths of + the backed up files relative to the backup root. The values in this + dictionary are the dates on which the particular file was backed up. """ if backup_name not in self.backups_dict: @@ -114,7 +118,7 @@ def get_backup_files(self, backup_name, original_paths=False): raise HedFileError("NoBackup", f"{backup_name} is not a valid backup", "") if original_paths: return [os.path.realpath(os.path.join(self.data_root, backup_key)) for backup_key in backup_dict.keys()] - return [os.path.realpath(os.path.join(self.backups_root, backup_name, self.BACKUP_ROOT, backup_key)) + return [os.path.realpath(os.path.join(self.backups_path, backup_name, self.BACKUP_ROOT, backup_key)) for backup_key in backup_dict.keys()] def get_backup_path(self, backup_name, file_name): @@ -128,7 +132,7 @@ def get_backup_path(self, backup_name, file_name): str: Full path of the corresponding file in the backup. """ - return os.path.realpath(os.path.join(self.backups_root, backup_name, self.BACKUP_ROOT, + return os.path.realpath(os.path.join(self.backups_path, backup_name, self.BACKUP_ROOT, self.get_file_key(file_name))) def get_file_key(self, file_name): @@ -163,8 +167,8 @@ def _get_backups(self): HedFileError - if a backup is inconsistent for any reason. """ backups = {} - for backup in os.listdir(self.backups_root): - backup_root = os.path.realpath(os.path.join(self.backups_root, backup)) + for backup in os.listdir(self.backups_path): + backup_root = os.path.realpath(os.path.join(self.backups_path, backup)) if not os.path.isdir(backup_root): raise HedFileError('BadBackupPath', f"{backup_root} is not a backup directory.", "") if len(os.listdir(backup_root)) != 2: @@ -195,12 +199,12 @@ def _check_backup_consistency(self, backup_name): """ - backup_dict_path = os.path.realpath(os.path.join(self.backups_root, backup_name, self.BACKUP_DICTIONARY)) + backup_dict_path = os.path.realpath(os.path.join(self.backups_path, backup_name, self.BACKUP_DICTIONARY)) if not os.path.exists(backup_dict_path): raise HedFileError("BadBackupDictionaryPath", f"Backup dictionary path {backup_dict_path} for backup " f"{backup_name} does not exist so backup invalid", "") - backup_root_path = os.path.realpath(os.path.join(self.backups_root, backup_name, self.BACKUP_ROOT)) + backup_root_path = os.path.realpath(os.path.join(self.backups_path, backup_name, self.BACKUP_ROOT)) if not os.path.isdir(backup_root_path): raise HedFileError("BadBackupRootPath", f"Backup root path {backup_root_path} for {backup_name} " diff --git a/hed/tools/remodeling/cli/run_remodel.py b/hed/tools/remodeling/cli/run_remodel.py index 9e044bb8a..aa5f8aa82 100644 --- a/hed/tools/remodeling/cli/run_remodel.py +++ b/hed/tools/remodeling/cli/run_remodel.py @@ -12,10 +12,10 @@ def get_parser(): """ Create a parser for the run_remodel command-line arguments. - + Returns: argparse.ArgumentParser: A parser for parsing the command line arguments. - + """ parser = argparse.ArgumentParser(description="Converts event files based on a json file specifying operations.") parser.add_argument("data_dir", help="Full path of dataset root directory.") @@ -33,6 +33,8 @@ def get_parser(): help="Optional path to JSON sidecar with HED information") parser.add_argument("-n", "--backup-name", default=BackupManager.DEFAULT_BACKUP_NAME, dest="backup_name", help="Name of the default backup for remodeling") + parser.add_argument("-nb", "--no-backup", action='store_true', dest="no_backup", + help="If present, the operations are run directly on the files with no backup.") parser.add_argument("-r", "--hed-versions", dest="hed_versions", nargs="*", default=[], help="Optional list of HED schema versions used for annotation, include prefixes.") parser.add_argument("-s", "--save-formats", nargs="*", default=['.json', '.txt'], dest="save_formats", @@ -40,6 +42,8 @@ def get_parser(): parser.add_argument("-t", "--task-names", dest="task_names", nargs="*", default=[], help="The names of the task.") parser.add_argument("-v", "--verbose", action='store_true', help="If present, output informative messages as computation progresses.") + parser.add_argument("-w", "--work-dir", default="", dest="work_dir", + help="If given, is the path to directory for saving, otherwise derivatives/remodel is used.") parser.add_argument("-x", "--exclude-dirs", nargs="*", default=[], dest="exclude_dirs", help="Directories names to exclude from search for files.") return parser @@ -148,19 +152,24 @@ def main(arg_list=None): args, operations = parse_arguments(arg_list) if not os.path.isdir(args.data_dir): raise HedFileError("DataDirectoryDoesNotExist", f"The root data directory {args.data_dir} does not exist", "") - if args.backup_name: + if args.no_backup: + backup_name = None + else: backup_man = BackupManager(args.data_dir) if not backup_man.get_backup(args.backup_name): raise HedFileError("BackupDoesNotExist", f"Backup {args.backup_name} does not exist. " f"Please run_remodel_backup first", "") backup_man.restore_backup(args.backup_name, args.task_names, verbose=args.verbose) - dispatch = Dispatcher(operations, data_root=args.data_dir, backup_name=args.backup_name, - hed_versions=args.hed_versions) + backup_name = args.backup_name + dispatch = Dispatcher(operations, data_root=args.data_dir, backup_name=backup_name, hed_versions=args.hed_versions) if args.use_bids: run_bids_ops(dispatch, args) else: run_direct_ops(dispatch, args) - dispatch.save_summaries(args.save_formats, individual_summaries=args.individual_summaries) + save_dir = None + if args.work_dir: + save_dir = os.path.realpath(os.path.join(args.work_dir, Dispatcher.REMODELING_SUMMARY_PATH)) + dispatch.save_summaries(args.save_formats, individual_summaries=args.individual_summaries, summary_dir=save_dir) if __name__ == '__main__': diff --git a/hed/tools/remodeling/cli/run_remodel_backup.py b/hed/tools/remodeling/cli/run_remodel_backup.py index c0e3e681a..40664f138 100644 --- a/hed/tools/remodeling/cli/run_remodel_backup.py +++ b/hed/tools/remodeling/cli/run_remodel_backup.py @@ -21,13 +21,19 @@ def get_parser(): help="Filename suffix of files to be backed up. A * indicates all files allowed.") parser.add_argument("-n", "--backup_name", default=BackupManager.DEFAULT_BACKUP_NAME, dest="backup_name", help="Name of the default backup for remodeling") + parser.add_argument("-p", "--path-work", default="", dest="path_work", + help="The root path for remodeling work if given, " + + "otherwise [data_root]/derivatives/remodel is used.") parser.add_argument("-t", "--task-names", dest="task_names", nargs="*", default=[], help="The name of the task.") parser.add_argument("-v", "--verbose", action='store_true', help="If present, output informative messages as computation progresses.") + parser.add_argument("-w", "--work-dir", default="", dest="work_dir", + help="If given, is the path to directory for saving, " + + "otherwise [data_root]derivatives/remodel is used.") parser.add_argument("-x", "--exclude-dirs", nargs="*", default=['derivatives'], dest="exclude_dirs", help="Directories names to exclude from search for files. " + "If omitted, no directories except the backup directory will be excluded." + - "Note data_dir/remodel/backup will always be excluded.") + "Note [data_root]/derivatives/remodel will always be excluded.") return parser @@ -55,7 +61,11 @@ def main(arg_list=None): exclude_dirs=exclude_dirs) if args.task_names: file_list = get_filtered_by_element(file_list, args.task_names) - backup_man = BackupManager(args.data_dir) + if args.work_dir: + backups_root = args.work_dir + else: + backups_root = None + backup_man = BackupManager(args.data_dir, backups_root=backups_root) if backup_man.get_backup(args.backup_name): raise HedFileError("BackupExists", f"Backup {args.backup_name} already exists", "") else: diff --git a/hed/tools/remodeling/cli/run_remodel_restore.py b/hed/tools/remodeling/cli/run_remodel_restore.py index b6da5d9db..79b136805 100644 --- a/hed/tools/remodeling/cli/run_remodel_restore.py +++ b/hed/tools/remodeling/cli/run_remodel_restore.py @@ -16,9 +16,13 @@ def get_parser(): parser.add_argument("data_dir", help="Full path of dataset root directory.") parser.add_argument("-n", "--backup_name", default=BackupManager.DEFAULT_BACKUP_NAME, dest="backup_name", help="Name of the default backup for remodeling") + parser.add_argument("-t", "--task-names", dest="task_names", nargs="*", default=[], help="The names of the task.") parser.add_argument("-v", "--verbose", action='store_true', help="If present, output informative messages as computation progresses.") + parser.add_argument("-w", "--work_dir", default="", dest="work_dir", + help="The root path for remodeling work if given, " + + "otherwise [data_root]/derivatives/remodel is used.") return parser @@ -36,7 +40,11 @@ def main(arg_list=None): """ parser = get_parser() args = parser.parse_args(arg_list) - backup_man = BackupManager(args.data_dir) + if args.work_dir: + backups_root = args.work_dir + else: + backups_root = None + backup_man = BackupManager(args.data_dir, backups_root=backups_root) if not backup_man.get_backup(args.backup_name): raise HedFileError("BackupDoesNotExist", f"{args.backup_name}", "") backup_man.restore_backup(args.backup_name, task_names=args.task_names, verbose=args.verbose) diff --git a/hed/tools/remodeling/dispatcher.py b/hed/tools/remodeling/dispatcher.py index 5371bb2d1..59d467226 100644 --- a/hed/tools/remodeling/dispatcher.py +++ b/hed/tools/remodeling/dispatcher.py @@ -14,7 +14,7 @@ class Dispatcher: """ Controller for applying operations to tabular files and saving the results. """ - REMODELING_SUMMARY_PATH = 'derivatives/remodel/summaries' + REMODELING_SUMMARY_PATH = 'remodel/summaries' def __init__(self, operation_list, data_root=None, backup_name=BackupManager.DEFAULT_BACKUP_NAME, hed_versions=None): @@ -128,7 +128,7 @@ def get_summary_save_dir(self): """ if self.data_root: - return os.path.realpath(os.path.join(self.data_root, Dispatcher.REMODELING_SUMMARY_PATH)) + return os.path.realpath(os.path.join(self.data_root, 'derivatives', Dispatcher.REMODELING_SUMMARY_PATH)) raise HedFileError("NoDataRoot", f"Dispatcher must have a data root to produce directories", "") def run_operations(self, file_path, sidecar=None, verbose=False): @@ -160,9 +160,10 @@ def save_summaries(self, save_formats=['.json', '.txt'], individual_summaries="s Parameters: save_formats (list): A list of formats [".txt", ."json"] individual_summaries (str): If True, include summaries of individual files. - summary_dir (str or None): Directory for saving summaries + summary_dir (str or None): Directory for saving summaries. - The summaries are saved in the dataset derivatives/remodeling folder if no save_dir is provided. + Notes: + The summaries are saved in the dataset derivatives/remodeling folder if no save_dir is provided. """ if not save_formats: diff --git a/hed/tools/remodeling/operations/base_context.py b/hed/tools/remodeling/operations/base_context.py index 1815ebd0d..a7a29d356 100644 --- a/hed/tools/remodeling/operations/base_context.py +++ b/hed/tools/remodeling/operations/base_context.py @@ -62,12 +62,12 @@ def get_summary(self, individual_summaries="separate"): Returns: dict - dictionary with "Dataset" and "Individual files" keys. - + Notes: The individual_summaries value is processed as follows - "separate" individual summaries are to be in separate files - "consolidated" means that the individual summaries are in same file as overall summary - "none" means that only the overall summary is produced. - + """ include_individual = individual_summaries == "separate" or individual_summaries == "consolidated" summary_details = self.get_summary_details(include_individual=include_individual) diff --git a/hed/tools/remodeling/operations/summarize_definitions_op.py b/hed/tools/remodeling/operations/summarize_definitions_op.py index 45fda3d82..a49c5d5dd 100644 --- a/hed/tools/remodeling/operations/summarize_definitions_op.py +++ b/hed/tools/remodeling/operations/summarize_definitions_op.py @@ -105,6 +105,15 @@ def update_context(self, new_context): self.errors.update(errors) def _get_details_dict(self, summary): + """ Return the summary-specific information in a dictionary. + + Parameters: + summary (?): Contains the resolved dictionaries. + + Returns: + dict: dictionary with the summary results. + + """ return None def _merge_all(self): diff --git a/hed/tools/remodeling/operations/summarize_hed_tags_op.py b/hed/tools/remodeling/operations/summarize_hed_tags_op.py index 89e494338..faa259e7f 100644 --- a/hed/tools/remodeling/operations/summarize_hed_tags_op.py +++ b/hed/tools/remodeling/operations/summarize_hed_tags_op.py @@ -117,6 +117,15 @@ def update_context(self, new_context): self.summary_dict[new_context["name"]] = counts def _get_details_dict(self, merge_counts): + """ Return the summary-specific information in a dictionary. + + Parameters: + merge_counts (HedTagCounts): Contains the counts of tags in the dataset. + + Returns: + dict: dictionary with the summary results. + + """ template, unmatched = merge_counts.organize_tags(self.tags) details = {} for key, key_list in self.tags.items(): diff --git a/hed/tools/remodeling/operations/summarize_hed_type_op.py b/hed/tools/remodeling/operations/summarize_hed_type_op.py index 7b3993357..fd500b4bc 100644 --- a/hed/tools/remodeling/operations/summarize_hed_type_op.py +++ b/hed/tools/remodeling/operations/summarize_hed_type_op.py @@ -113,6 +113,15 @@ def update_context(self, new_context): self.summary_dict[new_context["name"]] = counts def _get_details_dict(self, counts): + """ Return the summary-specific information in a dictionary. + + Parameters: + counts (HedTypeCounts): Contains the counts of the events in which the type occurs. + + Returns: + dict: dictionary with the summary results. + + """ return counts.get_summary() def _merge_all(self): diff --git a/tests/tools/remodeling/cli/test_run_remodel.py b/tests/tools/remodeling/cli/test_run_remodel.py index fac00f06b..893794e45 100644 --- a/tests/tools/remodeling/cli/test_run_remodel.py +++ b/tests/tools/remodeling/cli/test_run_remodel.py @@ -36,6 +36,9 @@ def setUp(self): def tearDown(self): shutil.rmtree(self.data_root) + work_path = os.path.realpath(os.path.join(self.extract_path, 'temp')) + if os.path.exists(work_path): + shutil.rmtree(work_path) @classmethod def tearDownClass(cls): @@ -73,6 +76,15 @@ def test_main_bids(self): main(arg_list) self.assertFalse(fp.getvalue()) + def test_main_bids_alt_path(self): + work_path = os.path.realpath(os.path.join(self.extract_path, 'temp')) + arg_list = [self.data_root, self.summary_model_path, '-x', 'derivatives', 'stimuli', '-r', '8.1.0', + '-j', self.sidecar_path, '-w', work_path] + + with patch('sys.stdout', new=io.StringIO()) as fp: + main(arg_list) + self.assertFalse(fp.getvalue()) + def test_main_bids_verbose_bad_task(self): arg_list = [self.data_root, self.model_path, '-x', 'derivatives', 'stimuli', '-b', '-t', 'junk', '-v'] with patch('sys.stdout', new=io.StringIO()) as fp: diff --git a/tests/tools/remodeling/cli/test_run_remodel_backup.py b/tests/tools/remodeling/cli/test_run_remodel_backup.py index a5a32e7ed..65f2391b6 100644 --- a/tests/tools/remodeling/cli/test_run_remodel_backup.py +++ b/tests/tools/remodeling/cli/test_run_remodel_backup.py @@ -47,7 +47,7 @@ def test_main_events(self): '-f', 'events', '-e', '.tsv'] main(arg_list) self.assertTrue(os.path.exists(self.derv_path), 'backup directory exists before creation') - json_path = os.path.realpath(os.path.join(self.derv_path, BackupManager.DEFAULT_BACKUP_NAME, + json_path = os.path.realpath(os.path.join(self.derv_path, 'backups', BackupManager.DEFAULT_BACKUP_NAME, BackupManager.BACKUP_DICTIONARY)) with open(json_path, 'r') as fp: key_dict = json.load(fp) @@ -62,12 +62,13 @@ def test_main_all(self): self.assertFalse(os.path.exists(self.derv_path), 'backup directory does not exist before creation') main(arg_list) self.assertTrue(os.path.exists(self.derv_path), 'backup directory exists before creation') - json_path = os.path.realpath(os.path.join(self.derv_path, BackupManager.DEFAULT_BACKUP_NAME, + json_path = os.path.realpath(os.path.join(self.derv_path, 'backups', BackupManager.DEFAULT_BACKUP_NAME, BackupManager.BACKUP_DICTIONARY)) with open(json_path, 'r') as fp: key_dict = json.load(fp) self.assertEqual(len(key_dict), 4, "The backup of events.tsv does not include top_level.tsv") - back_path = os.path.realpath(os.path.join(self.derv_path, BackupManager.DEFAULT_BACKUP_NAME, 'backup_root')) + back_path = os.path.realpath(os.path.join(self.derv_path, 'backups', + BackupManager.DEFAULT_BACKUP_NAME, 'backup_root')) file_list1 = get_file_list(back_path) self.assertIsInstance(file_list1, list) self.assertEqual(len(file_list1), 4) @@ -81,7 +82,7 @@ def test_main_task(self): '-f', 'events', '-e', '.tsv', '-t', 'FacePerception'] main(arg_list) self.assertTrue(os.path.exists(der_path)) - back_path = os.path.realpath(os.path.join(self.data_root, BackupManager.RELATIVE_BACKUP_LOCATION, + back_path = os.path.realpath(os.path.join(self.data_root, BackupManager.RELATIVE_BACKUP_LOCATION, 'backups', BackupManager.DEFAULT_BACKUP_NAME, 'backup_root')) self.assertTrue(os.path.exists(back_path)) backed_files = get_file_list(back_path) @@ -96,12 +97,29 @@ def test_main_bad_task(self): '-f', 'events', '-e', '.tsv', '-t', 'Baloney'] main(arg_list) self.assertTrue(os.path.exists(der_path)) - back_path = os.path.realpath(os.path.join(self.data_root, BackupManager.RELATIVE_BACKUP_LOCATION, + back_path = os.path.realpath(os.path.join(self.data_root, BackupManager.RELATIVE_BACKUP_LOCATION, 'backups', BackupManager.DEFAULT_BACKUP_NAME, 'backup_root')) self.assertTrue(os.path.exists(back_path)) backed_files = get_file_list(back_path) self.assertEqual(len(backed_files), 0) + def test_alt_loc(self): + alt_path = os.path.realpath(os.path.join(self.extract_path, 'temp')) + if os.path.exists(alt_path): + shutil.rmtree(alt_path) + self.assertFalse(os.path.exists(alt_path)) + arg_list = [self.data_root, '-n', BackupManager.DEFAULT_BACKUP_NAME, '-x', 'derivatives', '-w', alt_path, + '-f', 'events', '-e', '.tsv', ] + main(arg_list) + self.assertTrue(os.path.exists(alt_path)) + back_path = os.path.realpath(os.path.join(alt_path, 'backups', 'default_back', 'backup_root')) + self.assertTrue(os.path.exists(back_path)) + self.assertTrue(os.path.exists(back_path)) + backed_files = get_file_list(back_path) + self.assertEqual(len(backed_files), 6) + if os.path.exists(alt_path): + shutil.rmtree(alt_path) + def test_main_backup_exists(self): der_path = os.path.realpath(os.path.join(self.data_root, 'derivatives')) self.assertTrue(os.path.exists(der_path)) diff --git a/tests/tools/remodeling/cli/test_run_remodel_restore.py b/tests/tools/remodeling/cli/test_run_remodel_restore.py index e0a29dad2..c18dcbcfd 100644 --- a/tests/tools/remodeling/cli/test_run_remodel_restore.py +++ b/tests/tools/remodeling/cli/test_run_remodel_restore.py @@ -3,7 +3,9 @@ import unittest import zipfile from hed.errors import HedFileError +from hed.tools.remodeling.cli.run_remodel_backup import main as back_main from hed.tools.remodeling.cli.run_remodel_restore import main +from hed.tools.remodeling.backup_manager import BackupManager from hed.tools.util.io_util import get_file_list @@ -37,8 +39,7 @@ def test_main_restore(self): arg_list = [self.test_root_back1, '-n', 'back1'] main(arg_list) files3 = get_file_list(self.test_root_back1, exclude_dirs=['derivatives']) - overlap = set(files1).intersection(set(files3)) - self.assertEqual(len(overlap), len(files1), "run_restore restores all the files after") + self.assertEqual(len(files3), len(files1), "run_restore restores all the files after") def test_no_backup(self): # Test bad data directory @@ -47,6 +48,29 @@ def test_no_backup(self): main(arg_list=arg_list) self.assertEqual(context.exception.args[0], "BackupDoesNotExist") + def test_restore_alt_loc(self): + alt_path = os.path.realpath(os.path.join(self.extract_path, 'temp')) + if os.path.exists(alt_path): + shutil.rmtree(alt_path) + self.assertFalse(os.path.exists(alt_path)) + arg_list = [self.test_root_back1, '-n', 'back1', '-x', 'derivatives', '-w', alt_path, + '-f', 'events', '-e', '.tsv'] + back_main(arg_list) + files1 = get_file_list(self.test_root_back1, exclude_dirs=['derivatives']) + self.assertEqual(len(files1), 4, "run_restore starts with the right number of files.") + shutil.rmtree(os.path.realpath(os.path.join(self.test_root_back1, 'sub1'))) + shutil.rmtree(os.path.realpath(os.path.join(self.test_root_back1, 'sub2'))) + os.remove(os.path.realpath(os.path.join(self.test_root_back1, 'top_level.tsv'))) + files2 = get_file_list(self.test_root_back1, exclude_dirs=['derivatives']) + self.assertFalse(files2, "run_restore starts with the right number of files.") + arg_list = [self.test_root_back1, '-n', 'back1', '-w', alt_path,] + main(arg_list) + files3 = get_file_list(self.test_root_back1, exclude_dirs=['derivatives']) + self.assertEqual(len(files3)+1, len(files1), "run_restore restores all the files after") + + if os.path.exists(alt_path): + shutil.rmtree(alt_path) + if __name__ == '__main__': unittest.main() diff --git a/tests/tools/remodeling/test_backup_manager.py b/tests/tools/remodeling/test_backup_manager.py index 7b885d8a1..d441b5cc7 100644 --- a/tests/tools/remodeling/test_backup_manager.py +++ b/tests/tools/remodeling/test_backup_manager.py @@ -32,7 +32,7 @@ def setUpClass(cls): test_root_bad = os.path.realpath(os.path.join(os.path.dirname(__file__), '../../data/remodel_tests/test_root_bad')) cls.test_root_bad = test_root_bad - cls.test_root_bad_backups = os.path.join(test_root_bad, BackupManager.RELATIVE_BACKUP_LOCATION) + cls.test_root_bad_backups = os.path.join(test_root_bad, BackupManager.RELATIVE_BACKUP_LOCATION, 'backups') cls.test_paths_bad = [os.path.join(test_root_bad, file) for file in file_list] cls.test_zip_bad = os.path.realpath(os.path.join(os.path.dirname(__file__), '../../data/remodel_tests/test_root_bad.zip')) @@ -59,6 +59,21 @@ def test_constructor(self): self.assertIsInstance(back1_man, BackupManager, "constructor creates a BackupManager if no backups") self.assertTrue(back1_man.backups_dict) + def test_constructor_alternative_location(self): + alt_path = os.path.realpath(os.path.join(self.extract_path, 'temp_backs')) + back1_man = BackupManager(self.test_root_back1, backups_root=alt_path) + self.assertIsInstance(back1_man, BackupManager, "constructor creates a BackupManager if no backups") + self.assertFalse(back1_man.backups_dict) + file_list = get_file_list(self.test_root_back1, name_suffix='events', exclude_dirs=['derivatives'], + extensions=['.tsv']) + self.assertEqual(len(file_list), 3) + back1_man.create_backup(file_list, backup_name='my_back') + self.assertTrue(back1_man.backups_dict) + backup = back1_man.backups_dict['my_back'] + self.assertEqual(len(backup), len(file_list)) + if os.path.exists(alt_path): + shutil.rmtree(alt_path) + def test_bad_data_root(self): with self.assertRaises(HedFileError) as context: BackupManager('/baloney/Junk') @@ -126,5 +141,6 @@ def test_get_task(self): task3 = BackupManager.get_task(['abc', 'def'], 'temp/alpha_key_task_abc.txt') self.assertEqual(task3, 'abc') + if __name__ == '__main__': unittest.main() diff --git a/tests/tools/remodeling/test_dispatcher.py b/tests/tools/remodeling/test_dispatcher.py index 8fdd0ceaf..29329d9a9 100644 --- a/tests/tools/remodeling/test_dispatcher.py +++ b/tests/tools/remodeling/test_dispatcher.py @@ -85,7 +85,8 @@ def test_get_summary_save_dir(self): dispatch1 = Dispatcher(model1, data_root=self.test_root_back1, backup_name='back1') summary_path = dispatch1.get_summary_save_dir() self.assertEqual(summary_path, - os.path.realpath(os.path.join(self.test_root_back1, Dispatcher.REMODELING_SUMMARY_PATH))) + os.path.realpath(os.path.join(self.test_root_back1, 'derivatives', + Dispatcher.REMODELING_SUMMARY_PATH))) dispatch2 = Dispatcher(model1) with self.assertRaises(HedFileError) as context: dispatch2.get_summary_save_dir()