merged check_nested_run_dirs and check_nested_install_dirs

cylc/flow/workflow_files.py

@@ -1365,66 +1365,64 @@ def infer_latest_run(
     return (path, reg)
 
 
-def check_nested_run_path_base(path_: Union[Path, str]) -> None:
-    """Disallow nested installation directories:
+def check_nested_dirs(
+    path: Union[Path, str],
+    look_down: bool = False
+) -> None:
+    """Disallow nested dirs:
 
-    Do not allow installation into a directory whose parents and children
-    contain _cylc_install files.
+    - Nested installed run dirs
+    - Nested installed workflow dirs
 
     Args:
-        path_: The path to check
+        path: Absolute workflow run directory path or workflow install
+            directory path.
+        look_down: Search child dirs for recursion. (n.b. This is potentially
+            computationally expensive.)
 
     Raises:
-        WorkflowFilesError: If there is a ``_cylc-install`` directory
-        within maxdepth levels of path_
-    """
-    exc_msg = (
-        'Nested install directories not allowed - cannot install '
-        'workflow as "{0}" is already a valid workflow install directory.'
-    )
-    parent_dir = None
-
-    path_ = Path(path_)
-
-    globs = [
-        f'*/{"*/"*depth}_cylc-install' for depth in range(MAX_SCAN_DEPTH)
-    ]
-    for glob_ in globs:
-        if list(path_.glob(glob_)):
-            parent_dir = list(path_.glob(glob_))[0].parent
-    new_path = path_
-    for _ in range(1, MAX_SCAN_DEPTH):
-        new_path = new_path.parent
-        if list(new_path.glob('_cylc-install')):
-            parent_dir = list(new_path.glob('_cylc-install'))[0].parent
-
-    if parent_dir is not None and parent_dir.is_dir():
-        raise WorkflowFilesError(
-            exc_msg.format(get_cylc_run_abs_path(parent_dir))
-        )
-
-
-def check_nested_run_dirs(run_dir: Union[Path, str]) -> None:
-    """Disallow nested run dirs e.g. trying to install foo/bar where foo is
-    already a valid workflow directory.
-
-    Args:
-        run_dir: Absolute workflow run directory path.
-
-    Raises WorkflowFilesError if reg dir is nested inside a run dir.
+        WorkflowFilesError if reg dir is nested inside a run dir, or an
+            install dirs are nested.
     """
-    exc_msg = (
-        "Nested run directories not allowed - cannot install workflow in "
-        "'{0}' as '{1}' is already a valid run directory."
-    )
-    reg_path = Path(os.path.normpath(run_dir))
-    for parent_dir in reg_path.parents:
+    exc_msgs = {
+        'run_dir': (
+            "Nested run directories not allowed - cannot install workflow in "
+            "'{0}' as '{1}' is already a valid run directory."),
+        'install_dir': (
+            'Nested install directories not allowed - cannot install '
+            'workflow as "{0}" is already a valid workflow install directory.')
+    }
+    path = Path(os.path.normpath(path))
+
+    # Check parents:
+    for parent_dir in path.parents:
+        # Stop searching at ~/cylc-run
         if parent_dir == Path(get_cylc_run_dir()):
             break
+        # check for run directories:
         if is_valid_run_dir(parent_dir):
             raise WorkflowFilesError(
-                exc_msg.format(reg_path, get_cylc_run_abs_path(parent_dir))
-            )
+                exc_msgs['run_dir'].format(
+                    path, get_cylc_run_abs_path(parent_dir)))
+        # Check for install directories:
+        if (parent_dir / WorkflowFiles.Install.DIRNAME).is_dir():
+            raise WorkflowFilesError(
+                exc_msgs['install_dir'].format(
+                    get_cylc_run_abs_path(parent_dir)))
+
+    # Search child tree for install directories:
+    if look_down:
+        search_patterns = [
+            f'*/{"*/"*depth}{WorkflowFiles.Install.DIRNAME}'
+            for depth in range(MAX_SCAN_DEPTH)
+        ]
+        for search_pattern in search_patterns:
+            if list(path.glob(search_pattern)):
+                parent_dir = list(path.glob(search_pattern))[0].parent
+                raise WorkflowFilesError(
+                    exc_msgs['install_dir'].format(
+                        get_cylc_run_abs_path(parent_dir))
+                )
 
 
 def is_valid_run_dir(path):
@@ -1543,7 +1541,7 @@ def reinstall_workflow(named_run, rundir, source, dry_run=False):
             be changed.
     """
     validate_source_dir(source, named_run)
-    check_nested_run_dirs(rundir)
+    check_nested_dirs(rundir)
     reinstall_log = _get_logger(rundir, 'cylc-reinstall')
     reinstall_log.info(f"Reinstalling \"{named_run}\", from "
                        f"\"{source}\" to \"{rundir}\"")
@@ -1614,15 +1612,15 @@ def install_workflow(
         raise WorkflowFilesError(f'Run name cannot be "{run_name}".')
     validate_source_dir(source, flow_name)
     run_path_base = Path(get_workflow_run_dir(flow_name))
-    check_nested_run_path_base(run_path_base)
+    check_nested_dirs(run_path_base, look_down=True)
     relink, run_num, rundir = get_run_dir_info(
         run_path_base, run_name, no_run_name)
     if Path(rundir).exists():
         raise WorkflowFilesError(
             f"\"{rundir}\" exists."
             " Try using cylc reinstall. Alternatively, install with another"
             " name, using the --run-name option.")
-    check_nested_run_dirs(rundir)
+    check_nested_dirs(rundir)
     symlinks_created = {}
     named_run = flow_name
     if run_name:

tests/unit/test_workflow_files.py

@@ -36,12 +36,13 @@
 from cylc.flow.pathutil import parse_rm_dirs
 from cylc.flow.scripts.clean import CleanOptions
 from cylc.flow.workflow_files import (
+    MAX_SCAN_DEPTH,
     REG_CLASH_MSG,
     WorkflowFiles,
     _clean_using_glob,
     _remote_clean_cmd,
     check_flow_file,
-    check_nested_run_dirs,
+    check_nested_dirs,
     clean,
     get_rsync_rund_cmd,
     get_symlink_dirs,
@@ -101,25 +102,48 @@ def test_is_valid_run_dir(is_abs_path: bool, tmp_run_dir: Callable):
     assert workflow_files.is_valid_run_dir(Path(prefix, 'foo/bar')) is True
 
 
-def test_check_nested_run_dirs(tmp_run_dir: Callable):
-    """Test that check_nested_run_dirs() raises when a parent dir is a
+def test_check_nested_dirs(tmp_run_dir: Callable):
+    """Test that check_nested_dirs() raises when a parent dir is a
     workflow directory."""
     cylc_run_dir: Path = tmp_run_dir()
     test_dir = cylc_run_dir.joinpath('a/b/c/d/e')
     test_dir.mkdir(parents=True)
     # Parents are not run dirs - ok:
-    check_nested_run_dirs(test_dir)
+    check_nested_dirs(test_dir)
     # Parent contains a run dir but that run dir is not direct ancestor
     # of our test dir - ok:
     tmp_run_dir('a/Z')
-    check_nested_run_dirs(test_dir)
+    check_nested_dirs(test_dir)
     # Now make run dir out of parent - not ok:
     tmp_run_dir('a')
     with pytest.raises(WorkflowFilesError) as exc:
-        check_nested_run_dirs(test_dir)
+        check_nested_dirs(test_dir)
     assert "Nested run directories not allowed" in str(exc.value)
 
 
+@pytest.mark.parametrize(
+    'test_install_path, existing_install_path',
+[
+    (f'{"child/"*i}', '') for i in range(1,MAX_SCAN_DEPTH)
+] + [
+    ('', f'{"child/"*i}') for i in range(1,MAX_SCAN_DEPTH)
+]
+)
+def test_check_nested_dirs_install_dirs(
+    tmp_path: Path, test_install_path: str, existing_install_path: str
+):
+    """Test that check nested dirs looks both up and down a tree for
+    WorkflowFiles.Install.DIRNAME."""
+    testdir = tmp_path / test_install_path
+    testdir.mkdir(parents=True, exist_ok=True)
+    (tmp_path / existing_install_path / WorkflowFiles.Install.DIRNAME).mkdir(
+        parents=True
+    )
+    with pytest.raises(WorkflowFilesError) as exc:
+        check_nested_dirs(testdir, look_down=True)
+    assert "Nested install directories not allowed" in str(exc.value)
+
+
 @pytest.mark.parametrize(
     'reg, expected_err, expected_msg',
     [('foo/bar/', None, None),