diff --git a/src/aibs_informatics_core/utils/file_operations.py b/src/aibs_informatics_core/utils/file_operations.py index bd79c68..5d6eafd 100644 --- a/src/aibs_informatics_core/utils/file_operations.py +++ b/src/aibs_informatics_core/utils/file_operations.py @@ -182,6 +182,39 @@ def make_archive( raise ValueError(f"Error extracting file {source_path}. [{e}]") from e +def find_filesystem_boundary(starting_path: Path) -> Path: + """Given some starting Path, determine the nearest filesystem boundary (mount point). + If no mount is found, then this function will return the first parent directory PRIOR + to the filesystem anchor. + + >>> find_filesystem_boundary(Path("/allen/scratch/aibstemp")) + PosixPath('/allen/scratch') + + >>> find_filesystem_boundary(Path("/tmp/random_file.txt")) + PosixPath('/tmp') + + Args: + starting_path (Path): The starting Path + + Raises: + RuntimeError: If the provided starting_path cannot resolve to a real existing path + + Returns: + Path: The path of the nearest filesystem boundary OR the first parent directory prior + to the filesystem anchor (example anchors: "/", "c:\\") + """ + current_path = starting_path.resolve() + while current_path.parent != Path(current_path.anchor): + if current_path.is_mount(): + break + current_path = current_path.parent + + if current_path.exists(): + return current_path + else: + raise OSError(f"Could not find a real filesystem boundary for: {str(starting_path)}") + + def move_path(source_path: Path, destination_path: Path, exists_ok: bool = False): """Alias to simple mv command from one path to another diff --git a/test/aibs_informatics_core/utils/test_file_operations.py b/test/aibs_informatics_core/utils/test_file_operations.py index fecd308..5ef9516 100644 --- a/test/aibs_informatics_core/utils/test_file_operations.py +++ b/test/aibs_informatics_core/utils/test_file_operations.py @@ -15,6 +15,7 @@ PathLock, copy_path, extract_archive, + find_filesystem_boundary, get_path_hash, get_path_size_bytes, get_path_with_root, @@ -454,3 +455,34 @@ def test__strip_path_root__works(path, root, expected, expectation): actual = strip_path_root(path, root) if expected is not None: assert actual == expected + + +@mark.parametrize( + "starting_path, raise_expectation, expected", + [ + param( + # starting_path + Path("/tmp/test_subdir/test_file.txt"), + # raise_expectation + does_not_raise(), + # expected + Path("/tmp"), + id="Test basic case", + ), + param( + # starting_path + Path("/non_existent_dir/test_file.txt"), + # raise_expectation + raises(OSError, match=r"Could not find a real filesystem boundary.+"), + # expected + None, + id="Test non-existent path", + ), + ], +) +def test__find_filesystem_boundary(starting_path: Path, raise_expectation, expected: Path): + with raise_expectation: + obt = find_filesystem_boundary(starting_path=starting_path) + + if expected is not None: + assert expected == obt