Merge pull request #4166 from MetRonnie/dev

Tidy

cylc/flow/option_parsers.py

@@ -19,6 +19,7 @@
 from optparse import OptionParser, OptionConflictError, Values
 import os
 import sys
+from typing import Any, Dict, Optional
 
 from cylc.flow import LOG, RSYNC_LOG
 import cylc.flow.flags
@@ -273,7 +274,7 @@ def parse_args(self, api_args, remove_opts=None):
         return (options, args)
 
 
-class Options(Values):
+class Options:
     """Wrapper to allow Python API access to optparse CLI functionality.
 
     Example:
@@ -315,12 +316,14 @@ class Options(Values):
 
     """
 
-    def __init__(self, parser, overrides=None):
+    def __init__(
+        self, parser: OptionParser, overrides: Optional[Dict[str, Any]] = None
+    ) -> None:
         if overrides is None:
             overrides = {}
         self.defaults = {**parser.defaults, **overrides}
 
-    def __call__(self, **kwargs):
+    def __call__(self, **kwargs) -> Values:
         opts = Values(self.defaults)
         for key, value in kwargs.items():
             if hasattr(opts, key):

cylc/flow/pathutil.py

@@ -26,13 +26,15 @@
 from cylc.flow.exceptions import WorkflowFilesError
 from cylc.flow.platforms import (
     get_localhost_install_target,
-    get_platform
+    platform_from_name
 )
 
 
-def expand_path(path: Union[Path, str]) -> str:
-    """Expand both vars and user in path."""
-    return os.path.expanduser(os.path.expandvars(path))
+def expand_path(*args: Union[Path, str]) -> str:
+    """Expand both vars and user in path, joining any extra args."""
+    return os.path.expanduser(os.path.expandvars(
+        os.path.join(*args)
+    ))
 
 
 def get_remote_suite_run_dir(platform, suite, *args):
@@ -54,12 +56,17 @@ def get_remote_suite_work_dir(platform, suite, *args):
     )
 
 
-def get_workflow_run_dir(flow_name, *args):
+def get_workflow_run_dir(
+    flow_name: Union[Path, str], *args: Union[Path, str]
+) -> str:
     """Return local workflow run directory, joining any extra args, and
-    expanding vars and user."""
+    expanding vars and user.
+
+    Does not check that the directory exists.
+    """
     return expand_path(
         os.path.join(
-            get_platform()['run directory'], flow_name, *args
+            platform_from_name()['run directory'], flow_name, *args
         )
     )
 
@@ -97,14 +104,14 @@ def get_suite_run_pub_db_name(suite):
 def get_suite_run_share_dir(suite, *args):
     """Return local suite work/share directory, join any extra args."""
     return expand_path(os.path.join(
-        get_platform()['work directory'], suite, 'share', *args
+        platform_from_name()['work directory'], suite, 'share', *args
     ))
 
 
 def get_suite_run_work_dir(suite, *args):
     """Return local suite work/work directory, join any extra args."""
     return expand_path(os.path.join(
-        get_platform()['work directory'], suite, 'work', *args
+        platform_from_name()['work directory'], suite, 'work', *args
     ))
 
 

cylc/flow/scripts/clean.py

@@ -41,6 +41,11 @@
 from cylc.flow.terminal import cli_function
 from cylc.flow.suite_files import clean, init_clean
 
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from optparse import Values
+
 
 def get_option_parser():
     parser = COP(
@@ -65,7 +70,7 @@ def get_option_parser():
 
 
 @cli_function(get_option_parser)
-def main(parser, opts, reg):
+def main(parser: COP, opts: 'Values', reg: str):
     if not cylc.flow.flags.debug:
         # for readability omit timestamps from logging unless in debug mode
         for handler in LOG.handlers:

cylc/flow/scripts/hold.py

@@ -52,7 +52,7 @@
 from cylc.flow.terminal import cli_function
 
 if TYPE_CHECKING:
-    from cylc.flow.option_parsers import Options
+    from optparse import Values
 
 
 HOLD_MUTATION = '''
@@ -100,7 +100,7 @@ def get_option_parser() -> COP:
     return parser
 
 
-def _validate(options: 'Options', *task_globs: str) -> None:
+def _validate(options: 'Values', *task_globs: str) -> None:
     """Check combination of options and task globs is valid."""
     if options.hold_point_string:
         if task_globs:
@@ -115,7 +115,7 @@ def _validate(options: 'Options', *task_globs: str) -> None:
 
 
 @cli_function(get_option_parser)
-def main(parser: COP, options: 'Options', workflow: str, *task_globs: str):
+def main(parser: COP, options: 'Values', workflow: str, *task_globs: str):
 
     _validate(options, *task_globs)
 

cylc/flow/scripts/install.py

@@ -79,7 +79,7 @@
 from cylc.flow.terminal import cli_function
 
 if TYPE_CHECKING:
-    from cylc.flow.option_parsers import Options
+    from optparse import Values
 
 
 def get_option_parser():
@@ -173,7 +173,7 @@ def main(parser, opts, reg=None):
 
 
 def install(
-    parser: COP, opts: 'Options', reg: Optional[str] = None
+    parser: COP, opts: 'Values', reg: Optional[str] = None
 ) -> None:
     if opts.no_run_name and opts.run_name:
         parser.error(

cylc/flow/scripts/reinstall.py

@@ -51,7 +51,7 @@
 from cylc.flow.terminal import cli_function
 
 if TYPE_CHECKING:
-    from cylc.flow.option_parsers import Options
+    from optparse import Values
 
 
 def get_option_parser():
@@ -112,7 +112,7 @@ def get_option_parser():
 
 @cli_function(get_option_parser)
 def main(
-    parser: COP, opts: 'Options', named_run: Optional[str] = None
+    parser: COP, opts: 'Values', named_run: Optional[str] = None
 ) -> None:
     if not named_run:
         source, _ = get_workflow_source_dir(Path.cwd())

cylc/flow/scripts/release.py

@@ -47,7 +47,7 @@
 from cylc.flow.terminal import cli_function
 
 if TYPE_CHECKING:
-    from cylc.flow.option_parsers import Options
+    from optparse import Values
 
 
 RELEASE_MUTATION = '''
@@ -95,7 +95,7 @@ def get_option_parser() -> COP:
     return parser
 
 
-def _validate(options: 'Options', *task_globs: str) -> None:
+def _validate(options: 'Values', *task_globs: str) -> None:
     """Check combination of options and task globs is valid."""
     if options.release_all:
         if task_globs:
@@ -108,7 +108,7 @@ def _validate(options: 'Options', *task_globs: str) -> None:
 
 
 @cli_function(get_option_parser)
-def main(parser: COP, options: 'Options', workflow: str, *task_globs: str):
+def main(parser: COP, options: 'Values', workflow: str, *task_globs: str):
 
     _validate(options, *task_globs)
 

cylc/flow/suite_files.py

@@ -60,6 +60,7 @@
 from cylc.flow.wallclock import get_current_time_string
 
 if TYPE_CHECKING:
+    from optparse import Values
     from logging import Logger
 
 
@@ -592,7 +593,7 @@ def _clean_check(reg, run_dir):
             f"Cannot remove running workflow.\n\n{exc}")
 
 
-def init_clean(reg, opts):
+def init_clean(reg: str, opts: 'Values') -> None:
     """Initiate the process of removing a stopped workflow from the local
     scheduler filesystem and remote hosts.
 
@@ -883,13 +884,13 @@ def validate_flow_name(flow_name: str) -> None:
             f"workflow name cannot be an absolute path: {flow_name}")
 
 
-def check_nested_run_dirs(run_dir, flow_name):
+def check_nested_run_dirs(run_dir: Union[Path, str], flow_name: str) -> None:
     """Disallow nested run dirs e.g. trying to install foo/bar where foo is
     already a valid workflow directory.
 
     Args:
-        run_dir (path): run directory path
-        flow_name (str): workflow name
+        run_dir: Absolute workflow run directory path.
+        flow_name: Workflow name.
 
     Raise:
         WorkflowFilesError:
@@ -899,24 +900,25 @@ def check_nested_run_dirs(run_dir, flow_name):
     """
     exc_msg = (
         'Nested run directories not allowed - cannot install workflow name '
-        '"%s" as "%s" is already a valid run directory.')
+        '"{0}" as "{1}" is already a valid run directory.')
 
-    def _check_child_dirs(path, depth_count=1):
+    def _check_child_dirs(path: Union[Path, str], depth_count: int = 1):
         for result in os.scandir(path):
             if result.is_dir() and not result.is_symlink():
                 if is_valid_run_dir(result.path):
                     raise WorkflowFilesError(
-                        exc_msg %
-                        (flow_name, result.path))
+                        exc_msg.format(flow_name, result.path)
+                    )
                 if depth_count < MAX_SCAN_DEPTH:
                     _check_child_dirs(result.path, depth_count + 1)
 
-    reg_path = os.path.normpath(run_dir)
+    reg_path: Union[Path, str] = os.path.normpath(run_dir)
     parent_dir = os.path.dirname(reg_path)
     while parent_dir not in ['', '/']:
         if is_valid_run_dir(parent_dir):
             raise WorkflowFilesError(
-                exc_msg % (parent_dir, get_cylc_run_abs_path(parent_dir)))
+                exc_msg.format(parent_dir, get_cylc_run_abs_path(parent_dir))
+            )
         parent_dir = os.path.dirname(parent_dir)
 
     reg_path = get_cylc_run_abs_path(reg_path)
@@ -937,7 +939,7 @@ def is_valid_run_dir(path):
     return False
 
 
-def get_cylc_run_abs_path(path):
+def get_cylc_run_abs_path(path: Union[Path, str]) -> Union[Path, str]:
     """Return the absolute path under the cylc-run directory for the specified
     relative path.
 
@@ -1117,11 +1119,11 @@ def install_workflow(
         flow_name = Path.cwd().stem
     validate_flow_name(flow_name)
     if run_name in SuiteFiles.RESERVED_NAMES:
-        raise WorkflowFilesError(
-            f'Run name cannot be "{run_name}".')
+        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))
-    relink, run_num, rundir = get_run_dir(run_path_base, run_name, no_run_name)
+    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."
@@ -1132,7 +1134,7 @@ def install_workflow(
     if not no_symlinks:
         sub_dir = flow_name
         if run_num:
-            sub_dir += '/' + f'run{run_num}'
+            sub_dir = os.path.join(sub_dir, f'run{run_num}')
         symlinks_created = make_localhost_symlinks(rundir, sub_dir)
     INSTALL_LOG = _get_logger(rundir, 'cylc-install')
     if not no_symlinks and bool(symlinks_created) is True:
@@ -1177,28 +1179,24 @@ def install_workflow(
     return source, rundir, flow_name
 
 
-def get_run_dir(run_path_base, run_name, no_run_name):
-    """ Build run directory for current install.
+def get_run_dir_info(
+    run_path_base: Path, run_name: Optional[str], no_run_name: bool
+) -> Tuple[bool, Optional[int], Path]:
+    """Get (numbered, named or unnamed) run directory info for current install.
 
     Args:
-        run_path_base (Path):
-            The workflow directory.
-        run_name (str):
-            Name of the run.
-        no_run_name (bool):
-            Flag as True to indicate no run name - workflow installed into
-            ~/cylc-run/<run_path_base>.
+        run_path_base: The workflow directory absolute path.
+        run_name: Name of the run.
+        no_run_name: Flag as True to indicate no run name - workflow installed
+            into ~/cylc-run/<run_path_base>.
 
     Returns:
-        relink (bool):
-            True if runN symlink needs updating.
-        run_num (int):
-            Run number of the current install.
-        rundir (Path):
-            Run directory.
+        relink: True if runN symlink needs updating.
+        run_num: Run number of the current install, if using numbered runs.
+        rundir: Run directory absolute path.
     """
     relink = False
-    run_num = 0
+    run_num = None
     if no_run_name:
         rundir = run_path_base
     elif run_name:
@@ -1209,14 +1207,13 @@ def get_run_dir(run_path_base, run_name, no_run_name):
                 f"This path: \"{run_path_base}\" contains installed numbered"
                 " runs. Try again, using cylc install without --run-name.")
     else:
-        run_n = Path(run_path_base, SuiteFiles.RUN_N).expanduser()
         run_num = get_next_rundir_number(run_path_base)
         rundir = Path(run_path_base, f'run{run_num}')
         if run_path_base.exists() and detect_flow_exists(run_path_base, False):
             raise WorkflowFilesError(
                 f"This path: \"{run_path_base}\" contains an installed"
                 " workflow. Try again, using --run-name.")
-        unlink_runN(run_n)
+        unlink_runN(run_path_base)
         relink = True
     return relink, run_num, rundir
 
@@ -1246,7 +1243,8 @@ def detect_flow_exists(run_path_base, numbered):
 
 
 def check_flow_file(
-    path: Union[Path, str], symlink_suiterc: bool = False,
+    path: Union[Path, str],
+    symlink_suiterc: bool = False,
     logger: 'Logger' = LOG
 ) -> Path:
     """Raises WorkflowFilesError if no flow file in path sent.
@@ -1308,15 +1306,20 @@ def validate_source_dir(source, flow_name):
     check_flow_file(source)
 
 
-def unlink_runN(run_n):
-    """Remove symlink runN"""
+def unlink_runN(path: Union[Path, str]) -> bool:
+    """Remove symlink runN if it exists.
+
+    Args:
+        path: Absolute path to workflow dir containing runN.
+    """
     try:
-        Path(run_n).unlink()
+        Path(expand_path(path, SuiteFiles.RUN_N)).unlink()
     except OSError:
-        pass
+        return False
+    return True
 
 
-def link_runN(latest_run):
+def link_runN(latest_run: Union[Path, str]):
     """Create symlink runN, pointing at the latest run"""
     latest_run = Path(latest_run).expanduser()
     run_n = Path(latest_run.parent, SuiteFiles.RUN_N)