Add function generating unused directories given a list of paths

The create_unused_dir function distinguishes between the parent
directory and the name of the target subdirectory in 2 separate
arguments. This interface requires additional parameter management when
combined with newer path management interfaces such as pathlib.

- Add a function create_unused_dirs accepting the target paths as a
  single argument.
- Support the simultaneous creation of multiple unused paths with all of
  them having the same postfix if any of the actually requested paths is
  unavailable.

Author: gkaf89

easybuild/tools/filetools.py

@@ -2987,3 +2987,81 @@ def create_unused_dir(parent_folder, name):
     set_gid_sticky_bits(path, recursive=True)
 
     return path
+
+
+def get_first_nonexisting_parent(path):
+    path = os.path.abspath(path)
+
+    first_nonexisting_parent = None
+    while not os.path.exists(path):
+        first_nonexisting_parent = path
+        path = os.path.dirname(path)
+
+    return first_nonexisting_parent
+
+
+def _apply_postfix(paths, number):
+    postfix = '_' + str(number)
+
+    final_paths = []
+    first_nonexisting_parents = []
+    for p in paths:
+        final_paths.append(p + postfix)
+        first_nonexisting_parents.append(p)
+
+    return final_paths, first_nonexisting_parents
+
+
+def _clean_directories(paths):
+    for p in paths:
+        shutil.rmtree(p, ignore_errors=True)
+
+
+def create_unused_dirs(paths, index_upper_bound=10000):
+    """
+    Create directories with given paths, including the parent directories
+    When a directory in the same path for any of the path  in the path list already exists, then the suffix '_<i>' is
+    appended for i=0..(index_upper_bound-1) until an index is found so that all required path as unused. All created
+    directories have the same postfix.
+
+    :param paths: list of paths where directories are created; parent path created as well
+    :param index_upper_bound: maximum index that will be tried before failing
+    """
+    paths = list(map(os.path.abspath, paths))
+    for path in paths:
+        for parent in set(paths) - {path}:
+            if is_predecesor(parent, path):
+                raise EasyBuildError("Path '%s' is predecesor of '%s'" % (parent, path))
+
+    final_paths = paths
+    first_nonexisting_parents = list(map(get_first_nonexisting_parent, paths))
+    paths_exist = False
+    number = -1
+    while number < index_upper_bound and not paths_exist:  # Start with no suffix and limit the number of attempts
+        tried_paths = []
+        if number >= 0:
+            final_paths, first_nonexisting_parents = _apply_postfix(paths, number)
+        try:
+            for p in final_paths:
+                tried_paths.append(p)
+                os.makedirs(p)
+            paths_exist = True
+        except OSError as err:
+            # Distinguish between error due to existing folder and anything else
+            if tried_paths != [] and not os.path.exists(tried_paths[-1]):
+                _clean_directories(tried_paths[0:-1])
+                raise EasyBuildError("Failed to create directory %s: %s", tried_paths[-1], err)
+            _clean_directories(tried_paths[0:-1])
+        except BaseException as err:
+            _clean_directories(tried_paths)
+            raise err
+        number += 1
+
+    if number == index_upper_bound and not paths_exist:
+        raise EasyBuildError("Exceeded maximum number of attempts to generate unique directory: %s" % number)
+
+    # set group ID and sticky bits, if desired
+    for p in first_nonexisting_parents:
+        set_gid_sticky_bits(p, recursive=True)
+
+    return final_paths

test/framework/filetools.py

@@ -3544,6 +3544,31 @@ def test_compat_makedirs(self):
         py2vs3.makedirs(name, exist_ok=True)  # No error
         self.assertExists(name)
 
+    def test_get_first_nonexisting_parent(self):
+        """Test get_first_nonexisting_parent function."""
+        test_root = os.path.join(self.test_prefix, 'a')
+
+        base_path = os.path.join(self.test_prefix, 'a')
+        target_path = os.path.join(self.test_prefix, 'a', 'b', 'c')
+        os.makedirs(base_path)
+        first_nonexisting_parent = ft.get_first_nonexisting_parent(target_path)
+        self.assertEqual(first_nonexisting_parent, os.path.join(self.test_prefix, 'a', 'b'))
+        shutil.rmtree(test_root)
+
+        base_path = os.path.join(self.test_prefix, 'a', 'b')
+        target_path = os.path.join(self.test_prefix, 'a', 'b', 'c')
+        os.makedirs(base_path)
+        first_nonexisting_parent = ft.get_first_nonexisting_parent(target_path)
+        self.assertEqual(first_nonexisting_parent, os.path.join(self.test_prefix, 'a', 'b', 'c'))
+        shutil.rmtree(test_root)
+
+        base_path = os.path.join(self.test_prefix, 'a', 'b', 'c')
+        target_path = os.path.join(self.test_prefix, 'a', 'b', 'c')
+        os.makedirs(base_path)
+        first_nonexisting_parent = ft.get_first_nonexisting_parent(target_path)
+        self.assertEqual(first_nonexisting_parent, None)
+        shutil.rmtree(test_root)
+
     def test_create_unused_dir(self):
         """Test create_unused_dir function."""
         path = ft.create_unused_dir(self.test_prefix, 'folder')
@@ -3582,6 +3607,126 @@ def test_create_unused_dir(self):
         self.assertEqual(path, os.path.join(self.test_prefix, 'file_0'))
         self.assertExists(path)
 
+    def test_create_unused_dirs(self):
+        """Test create_unused_dirs function."""
+        requested_paths = []
+        requested_paths.append(os.path.join(self.test_prefix, 'folder_a'))
+        requested_paths.append(os.path.join(self.test_prefix, 'folder_b'))
+        paths = ft.create_unused_dirs(requested_paths)
+        self.assertEqual(paths, requested_paths)
+        map(lambda p: self.assertExists(p), paths)
+
+        # Repeat with existing folder(s) should create new ones
+        for i in range(10):
+            requested_paths = []
+            requested_paths.append(os.path.join(self.test_prefix, 'folder_a'))
+            requested_paths.append(os.path.join(self.test_prefix, 'folder_b'))
+            paths = ft.create_unused_dirs(requested_paths)
+            self.assertEqual(paths, list(map(lambda p: p + '_%s' % i, requested_paths)))
+            map(lambda p: self.assertExists(p), paths)
+
+        # Skip suffix if a directory with the suffix already exists
+        existing_idx = 1
+        os.mkdir(os.path.join(self.test_prefix, "existing_idx_a"))
+        os.mkdir(os.path.join(self.test_prefix, "existing_idx_b"))
+        os.mkdir(os.path.join(self.test_prefix, f"existing_idx_b_{existing_idx}"))
+
+        def expected_idx(n):
+            if n >= existing_idx:
+                return n + 1
+            return n
+
+        for i in range(3):
+            requested_paths = []
+            requested_paths.append(os.path.join(self.test_prefix, 'existing_idx_a'))
+            requested_paths.append(os.path.join(self.test_prefix, 'existing_idx_b'))
+            paths = ft.create_unused_dirs(requested_paths)
+            self.assertEqual(paths, list(map(lambda p: p + '_%s' % expected_idx(i), requested_paths)))
+            map(lambda p: self.assertExists(p), paths)
+            self.assertNotExists(os.path.join(self.test_prefix, f"existing_idx_a_{existing_idx}"))
+            self.assertExists(os.path.join(self.test_prefix, f"existing_idx_b_{existing_idx}"))
+
+        # Support creation of parent directories
+        requested_paths = [os.path.join(self.test_prefix, 'parent_folder', 'folder')]
+        paths = ft.create_unused_dirs(requested_paths)
+        self.assertEqual(paths, requested_paths)
+        map(lambda p: self.assertExists(p), paths)
+
+        # Not influenced by similar folder
+        requested_paths = [os.path.join(self.test_prefix, 'folder_a2')]
+        paths = ft.create_unused_dirs(requested_paths)
+        self.assertEqual(paths, requested_paths)
+        map(lambda p: self.assertExists(p), paths)
+        for i in range(10):
+            paths = ft.create_unused_dirs(requested_paths)
+            self.assertEqual(paths, list(map(lambda p: p + '_%s' % i, requested_paths)))
+            map(lambda p: self.assertExists(p), paths)
+
+        # Fail cleanly if passed a readonly folder
+        readonly_dir = os.path.join(self.test_prefix, 'ro_folder')
+        ft.mkdir(readonly_dir)
+        old_perms = os.lstat(readonly_dir)[stat.ST_MODE]
+        ft.adjust_permissions(readonly_dir, stat.S_IREAD | stat.S_IEXEC, relative=False)
+        requested_path = os.path.join(readonly_dir, 'new_folder')
+        try:
+            self.assertErrorRegex(EasyBuildError, 'Failed to create directory',
+                                  ft.create_unused_dirs, [requested_path])
+        finally:
+            ft.adjust_permissions(readonly_dir, old_perms, relative=False)
+
+        # Fail if the number of attempts to create the directory is exceeded
+        os.mkdir(os.path.join(self.test_prefix, 'attempt'))
+        os.mkdir(os.path.join(self.test_prefix, 'attempt_0'))
+        os.mkdir(os.path.join(self.test_prefix, 'attempt_1'))
+        os.mkdir(os.path.join(self.test_prefix, 'attempt_2'))
+        os.mkdir(os.path.join(self.test_prefix, 'attempt_3'))
+        requested_path = os.path.join(self.test_prefix, 'attempt')
+        self.assertErrorRegex(
+            EasyBuildError,
+            'Exceeded maximum number of attempts to generate unique directory',
+            ft.create_unused_dirs,
+            [requested_path], index_upper_bound=4
+        )
+
+        # Ignore files same as folders. So first just create a file with no contents
+        requested_path = os.path.join(self.test_prefix, 'file')
+        ft.write_file(requested_path, '')
+        paths = ft.create_unused_dirs([requested_path])
+        self.assertEqual(paths, [requested_path + '_0'])
+        map(lambda p: self.assertExists(p), paths)
+
+        # Deny creation of nested directories
+        requested_paths = [
+            os.path.join(self.test_prefix, 'nested_a/foo/bar'),
+            os.path.join(self.test_prefix, 'nested_a/foo/bar/baz'),
+        ]
+        self.assertErrorRegex(
+            EasyBuildError,
+            "Path '.*/foo/bar' is predecesor of '.*/foo/bar/baz'",
+            ft.create_unused_dirs,
+            requested_paths
+        )
+        requested_paths = [
+            os.path.join(self.test_prefix, 'nested_b/foo/bar/baz'),
+            os.path.join(self.test_prefix, 'nested_b/foo/bar'),
+        ]
+        self.assertErrorRegex(
+            EasyBuildError,
+            "Path '.*/foo/bar' is predecesor of '.*/foo/bar/baz'",
+            ft.create_unused_dirs,
+            requested_paths
+        )
+
+        # Allow creation of non-nested directories
+        requested_paths = [
+            os.path.join(self.test_prefix, 'nested_c/foo/bar'),
+            os.path.join(self.test_prefix, 'nested_c/foo/baz'),
+            os.path.join(self.test_prefix, 'nested_c/buz'),
+        ]
+        paths = ft.create_unused_dirs(requested_paths)
+        self.assertEqual(paths, requested_paths)
+        map(lambda p: self.assertExists(p), paths)
+
 
 def suite():
     """ returns all the testcases in this module """