WIP: --single-process inherit reqs & hints

Author: mr-c

cwltool/argparser.py

@@ -656,8 +656,10 @@ def arg_parser() -> argparse.ArgumentParser:
         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. "
+        "This will not include any step-level processing: 'scatter', 'when'; "
+        "and there will be no processing of step-level 'default', or 'valueFrom' "
+        "input modifiers. However, requirements/hints from the step or parent "
+        "workflow(s) will be inherited as usual."
         "The input object must match that Process's inputs.",
     )
 

cwltool/main.py

@@ -754,6 +754,32 @@ def my_represent_none(
     )
 
 
+def inherit_reqshints(tool: Process, parent: Process) -> None:
+    """Copy down requirements and hints from ancestors of a given process."""
+    for parent_req in parent.requirements:
+        found = False
+        for tool_req in tool.requirements:
+            if parent_req["class"] == tool_req["class"]:
+                found = True
+                break
+        if not found:
+            tool.requirements.append(parent_req)
+    for parent_hint in parent.hints:
+        found = False
+        for tool_req in tool.requirements:
+            if parent_hint["class"] == tool_req["class"]:
+                found = True
+                break
+        if not found:
+            for tool_hint in tool.hints:
+                if parent_hint["class"] == tool_hint["class"]:
+                    found = True
+                    break
+            if not found:
+                tool.hints.append(parent_req)
+    # TODO: get the parent of the parent and recurse
+
+
 def choose_target(
     args: argparse.Namespace,
     tool: Process,
@@ -826,36 +852,33 @@ def choose_process(
     tool: Process,
     loadingContext: LoadingContext,
 ) -> Optional[Process]:
-    """Walk the given Workflow and extract just args.single_step."""
+    """Walk the given Workflow and extract just args.single_process."""
     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,
-            )
+            step_id = tool.tool["id"] + "/" + args.single_process
         else:
-            extracted = get_process(
-                tool,
-                loadingContext.loader.fetcher.urljoin(
-                    tool.tool["id"], "#" + args.single_process
-                ),
-                loadingContext.loader.idx,
+            step_id = loadingContext.loader.fetcher.urljoin(
+                tool.tool["id"], "#" + args.single_process
             )
+        extracted, step_pos = get_process(
+            tool,
+            step_id,
+            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)
+        new_tool = make_tool(extracted["id"], loadingContext)
     else:
         raise Exception("Missing loadingContext.loader.idx!")
-
-    return tool
+    inherit_reqshints(new_tool, tool.steps[step_pos])
+    return new_tool
 
 
 def check_working_directories(

cwltool/subgraph.py

@@ -15,6 +15,7 @@
 
 from ruamel.yaml.comments import CommentedMap
 
+from .process import Process
 from .utils import CWLObjectType, aslist
 from .workflow import Workflow, WorkflowStep
 
@@ -55,11 +56,14 @@ def declare_node(nodes: Dict[str, Node], nodeid: str, tp: Optional[str]) -> Node
     return nodes[nodeid]
 
 
-def find_step(steps: List[WorkflowStep], stepid: str) -> Optional[CWLObjectType]:
-    for st in steps:
+def find_step(
+    steps: List[WorkflowStep], stepid: str
+) -> Tuple[Optional[CWLObjectType], int]:
+    """Find the step and step_index for a given step id."""
+    for index, st in enumerate(steps):
         if st.tool["id"] == stepid:
-            return st.tool
-    return None
+            return st.tool, index
+    return None, -1
 
 
 def get_subgraph(roots: MutableSequence[str], tool: Workflow) -> CommentedMap:
@@ -124,7 +128,7 @@ def get_subgraph(roots: MutableSequence[str], tool: Workflow) -> CommentedMap:
                     df = urllib.parse.urldefrag(u)
                     rn = str(df[0] + "#" + df[1].replace("/", "_"))
                     if nodes[v].type == STEP:
-                        wfstep = find_step(tool.steps, v)
+                        wfstep = find_step(tool.steps, v)[0]
                         if wfstep is not None:
                             for inp in cast(
                                 MutableSequence[CWLObjectType], wfstep["inputs"]
@@ -169,7 +173,7 @@ def get_step(tool: Workflow, step_id: str) -> CommentedMap:
 
     extracted = CommentedMap()
 
-    step = find_step(tool.steps, step_id)
+    step = find_step(tool.steps, step_id)[0]
     if step is None:
         raise Exception(f"Step {step_id} was not found")
 
@@ -197,15 +201,18 @@ def get_step(tool: Workflow, step_id: str) -> CommentedMap:
     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)
+def get_process(
+    tool: Workflow, step_id: str, index: Mapping[str, Any]
+) -> Tuple[Any, int]:
+    """Find the underlying Process for a given Workflow step id."""
+    step, step_pos = 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]
+        process = index[run]
     else:
-        return run
+        process = run
+    return process, step_pos

tests/subgraph/env-job.json

@@ -0,0 +1,3 @@
+{
+    "in": "hello test env"
+}

tests/subgraph/env-tool2.cwl

@@ -0,0 +1,18 @@
+class: CommandLineTool
+cwlVersion: v1.2
+inputs:
+  in: string
+outputs:
+  out:
+    type: File
+    outputBinding:
+      glob: out
+
+hints:
+  EnvVarRequirement:
+    envDef:
+      TEST_ENV: $(inputs.in)
+
+baseCommand: ["/bin/sh", "-c", "echo $TEST_ENV"]
+
+stdout: out

tests/subgraph/env-wf2.cwl

@@ -0,0 +1,23 @@
+#!/usr/bin/env cwl-runner
+class: Workflow
+cwlVersion: v1.2
+
+inputs:
+    in: string
+
+outputs:
+    out:
+      type: File
+      outputSource: step1/out
+
+requirements:
+  EnvVarRequirement:
+    envDef:
+      TEST_ENV: override
+
+steps:
+  step1:
+    run: env-tool2.cwl
+    in:
+      in: in
+    out: [out]

tests/test_subgraph.py

@@ -9,7 +9,7 @@
 from cwltool.subgraph import get_step, get_subgraph
 from cwltool.workflow import Workflow, default_make_tool
 
-from .util import get_data
+from .util import get_data, get_main_output
 
 
 def clean(val: Any, path: str) -> Any:
@@ -87,3 +87,20 @@ def test_get_step() -> None:
         extracted = get_step(tool, wf + "#" + a)
         with open(get_data("tests/subgraph/single_" + a + ".json")) as f:
             assert json.load(f) == clean(convert_to_dict(extracted), sg)
+
+
+def test_single_process_inherit_reqshints() -> None:
+    """Inherit reqs and hints from parent(s) with --single-step."""
+    err_code, stdout, stderr = get_main_output(
+        [
+            "--single-process",
+            "step1",
+            get_data("tests/subgraph/env-wf2.cwl"),
+            get_data("tests/subgraph/env-job.json"),
+        ]
+    )
+    assert err_code == 0
+    assert (
+        json.loads(stdout)["out"]["checksum"]
+        == "sha1$cdc1e84968261d6a7575b5305945471f8be199b6"
+    )