--single-step: fix typo ; add --single-process (#1444)

Author: mr-c

cwltool/argparser.py

@@ -644,6 +644,16 @@ def arg_parser() -> argparse.ArgumentParser:
         help="Only executes a single step in a workflow. The input object must "
         "match that step's inputs. Can be combined with --print-subgraph.",
     )
+    subgroup.add_argument(
+        "--single-process",
+        type=str,
+        default=None,
+        help="Only executes the underlying Process (CommandLineTool, "
+        "ExpressionTool, or sub-Workflow) for the given step in a workflow. "
+        "This will not include any step-level processing: scatter, when, no "
+        "processing of step-level default, or valueFrom input modifiers. "
+        "The input object must match that Process's inputs.",
+    )
 
     parser.add_argument(
         "--mpi-config-file",

cwltool/job.py

@@ -15,9 +15,9 @@
 from io import IOBase
 from threading import Timer
 from typing import (
+    IO,
     Callable,
     Dict,
-    IO,
     Iterable,
     List,
     Mapping,

cwltool/main.py

@@ -81,7 +81,7 @@
     get_container_from_software_requirements,
 )
 from .stdfsaccess import StdFsAccess
-from .subgraph import get_step, get_subgraph
+from .subgraph import get_process, get_step, get_subgraph
 from .update import ALLUPDATES, UPDATES
 from .utils import (
     DEFAULT_TMP_PREFIX,
@@ -811,7 +811,7 @@ def choose_step(
     if isinstance(tool, Workflow):
         url = urllib.parse.urlparse(tool.tool["id"])
         if url.fragment:
-            extracted = get_step(tool, tool.tool["id"] + "/" + args.singe_step)
+            extracted = get_step(tool, tool.tool["id"] + "/" + args.single_step)
         else:
             extracted = get_step(
                 tool,
@@ -831,6 +831,43 @@ def choose_step(
     return tool
 
 
+def choose_process(
+    args: argparse.Namespace,
+    tool: Process,
+    loadingContext: LoadingContext,
+) -> Optional[Process]:
+    """Walk the given Workflow and extract just args.single_step."""
+    if loadingContext.loader is None:
+        raise Exception("loadingContext.loader cannot be None")
+
+    if isinstance(tool, Workflow):
+        url = urllib.parse.urlparse(tool.tool["id"])
+        if url.fragment:
+            extracted = get_process(
+                tool,
+                tool.tool["id"] + "/" + args.single_process,
+                loadingContext.loader.idx,
+            )
+        else:
+            extracted = get_process(
+                tool,
+                loadingContext.loader.fetcher.urljoin(
+                    tool.tool["id"], "#" + args.single_process
+                ),
+                loadingContext.loader.idx,
+            )
+    else:
+        _logger.error("Can only use --single-process on Workflows")
+        return None
+    if isinstance(loadingContext.loader.idx, MutableMapping):
+        loadingContext.loader.idx[extracted["id"]] = extracted
+        tool = make_tool(extracted["id"], loadingContext)
+    else:
+        raise Exception("Missing loadingContext.loader.idx!")
+
+    return tool
+
+
 def check_working_directories(
     runtimeContext: RuntimeContext,
 ) -> Optional[int]:
@@ -1068,6 +1105,13 @@ def main(
                 else:
                     tool = ctool
 
+            elif args.single_process:
+                ctool = choose_process(args, tool, loadingContext)
+                if ctool is None:
+                    return 1
+                else:
+                    tool = ctool
+
             if args.print_subgraph:
                 if "name" in tool.tool:
                     del tool.tool["name"]

cwltool/subgraph.py

@@ -1,8 +1,10 @@
 import urllib
 from collections import namedtuple
 from typing import (
+    Any,
     Dict,
     List,
+    Mapping,
     MutableMapping,
     MutableSequence,
     Optional,
@@ -191,3 +193,17 @@ def get_step(tool: Workflow, step_id: str) -> CommentedMap:
             extracted[f] = tool.tool[f]
 
     return extracted
+
+
+def get_process(tool: Workflow, step_id: str, index: Mapping[str, Any]) -> Any:
+    """Return just a single Process from a Workflow step."""
+    step = find_step(tool.steps, step_id)
+    if step is None:
+        raise Exception(f"Step {step_id} was not found")
+
+    run = step["run"]
+
+    if isinstance(run, str):
+        return index[run]
+    else:
+        return run

tests/test_environment.py

@@ -1,12 +1,13 @@
 """Test passing of environment variables to tools."""
-from abc import ABC, abstractmethod
 import os
+from abc import ABC, abstractmethod
 from pathlib import Path
 from typing import Any, Callable, Dict, Iterator, List, Mapping, Set, Union, cast
 
 import pytest
 
 from cwltool.singularity import get_version
+
 from .util import env_accepts_null, get_tool_env, needs_docker, needs_singularity
 
 # None => accept anything, just require the key is present

tests/test_singularity.py

@@ -1,6 +1,6 @@
 """Tests to find local Singularity image."""
-import shutil
 import os
+import shutil
 from pathlib import Path
 from typing import Any
 

tests/util.py

@@ -5,8 +5,8 @@
 import os
 import shutil
 import subprocess
-from pathlib import Path
 import sys
+from pathlib import Path
 from typing import Dict, Generator, List, Mapping, Optional, Tuple, Union
 
 import pytest