print-targets/subgraph/single_*: load step process + recurse

and tests

Author: mr-c

cwltool/cwlviewer.py

@@ -1,5 +1,4 @@
 """Visualize a CWL workflow."""
-import os
 from pathlib import Path
 from urllib.parse import urlparse
 

cwltool/main.py

@@ -55,6 +55,7 @@
     fetch_document,
     jobloaderctx,
     load_overrides,
+    load_tool,
     make_tool,
     resolve_and_validate_document,
     resolve_overrides,
@@ -776,72 +777,76 @@ def inherit_reqshints(tool: Process, parent: Process) -> None:
                     found = True
                     break
             if not found:
-                tool.hints.append(parent_req)
+                tool.hints.append(parent_hint)
 
 
 def choose_target(
     args: argparse.Namespace,
     tool: Process,
-    loadingContext: LoadingContext,
+    loading_context: LoadingContext,
 ) -> Optional[Process]:
     """Walk the Workflow, extract the subset matches all the args.targets."""
-    if loadingContext.loader is None:
-        raise Exception("loadingContext.loader cannot be None")
+    if loading_context.loader is None:
+        raise Exception("loading_context.loader cannot be None")
 
     if isinstance(tool, Workflow):
         url = urllib.parse.urlparse(tool.tool["id"])
         if url.fragment:
             extracted = get_subgraph(
-                [tool.tool["id"] + "/" + r for r in args.target], tool
+                [tool.tool["id"] + "/" + r for r in args.target], tool, loading_context
             )
         else:
             extracted = get_subgraph(
                 [
-                    loadingContext.loader.fetcher.urljoin(tool.tool["id"], "#" + r)
+                    loading_context.loader.fetcher.urljoin(tool.tool["id"], "#" + r)
                     for r in args.target
                 ],
                 tool,
+                loading_context,
             )
     else:
         _logger.error("Can only use --target on Workflows")
         return None
-    if isinstance(loadingContext.loader.idx, MutableMapping):
-        loadingContext.loader.idx[extracted["id"]] = extracted
-        tool = make_tool(extracted["id"], loadingContext)
+    if isinstance(loading_context.loader.idx, MutableMapping):
+        loading_context.loader.idx[extracted["id"]] = extracted
+        tool = make_tool(extracted["id"], loading_context)
     else:
-        raise Exception("Missing loadingContext.loader.idx!")
+        raise Exception("Missing loading_context.loader.idx!")
 
     return tool
 
 
 def choose_step(
     args: argparse.Namespace,
     tool: Process,
-    loadingContext: LoadingContext,
+    loading_context: 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 loading_context.loader is None:
+        raise Exception("loading_context.loader cannot be None")
 
     if isinstance(tool, Workflow):
         url = urllib.parse.urlparse(tool.tool["id"])
         if url.fragment:
-            extracted = get_step(tool, tool.tool["id"] + "/" + args.single_step)
+            extracted = get_step(
+                tool, tool.tool["id"] + "/" + args.single_step, loading_context
+            )
         else:
             extracted = get_step(
                 tool,
-                loadingContext.loader.fetcher.urljoin(
+                loading_context.loader.fetcher.urljoin(
                     tool.tool["id"], "#" + args.single_step
                 ),
+                loading_context,
             )
     else:
         _logger.error("Can only use --single-step on Workflows")
         return None
-    if isinstance(loadingContext.loader.idx, MutableMapping):
-        loadingContext.loader.idx[extracted["id"]] = extracted
-        tool = make_tool(extracted["id"], loadingContext)
+    if isinstance(loading_context.loader.idx, MutableMapping):
+        loading_context.loader.idx[extracted["id"]] = extracted
+        tool = make_tool(extracted["id"], loading_context)
     else:
-        raise Exception("Missing loadingContext.loader.idx!")
+        raise Exception("Missing loading_context.loader.idx!")
 
     return tool
 
@@ -866,7 +871,7 @@ def choose_process(
         extracted, step_pos = get_process(
             tool,
             step_id,
-            loadingContext.loader.idx,
+            loadingContext,
         )
     else:
         _logger.error("Can only use --single-process on Workflows")
@@ -909,6 +914,33 @@ def check_working_directories(
     return None
 
 
+def print_targets(
+    tool: Process,
+    stdout: Union[TextIO, StreamWriter],
+    loading_context: LoadingContext,
+    prefix: str = "",
+) -> None:
+    """Recursively find targets for --subgraph and friends."""
+    for f in ("outputs", "inputs"):
+        if tool.tool[f]:
+            _logger.info("%s %s%s targets:", prefix[:-1], f[0].upper(), f[1:-1])
+            stdout.write(
+                "  "
+                + "\n  ".join([f"{prefix}{shortname(t['id'])}" for t in tool.tool[f]])
+                + "\n"
+            )
+    if "steps" in tool.tool:
+        _logger.info("%s steps targets:", prefix[:-1])
+        for t in tool.tool["steps"]:
+            stdout.write(f"  {prefix}{shortname(t['id'])}\n")
+            run: Union[str, Process] = t["run"]
+            if isinstance(run, str):
+                process = load_tool(run, loading_context)
+            else:
+                process = run
+            print_targets(process, stdout, loading_context, shortname(t["id"]) + "/")
+
+
 def main(
     argsl: Optional[List[str]] = None,
     args: Optional[argparse.Namespace] = None,
@@ -1106,14 +1138,7 @@ def main(
                 return 0
 
             if args.print_targets:
-                for f in ("outputs", "steps", "inputs"):
-                    if tool.tool[f]:
-                        _logger.info("%s%s targets:", f[0].upper(), f[1:-1])
-                        stdout.write(
-                            "  "
-                            + "\n  ".join([shortname(t["id"]) for t in tool.tool[f]])
-                            + "\n"
-                        )
+                print_targets(tool, stdout, loadingContext)
                 return 0
 
             if args.target:

cwltool/subgraph.py

@@ -10,11 +10,14 @@
     Optional,
     Set,
     Tuple,
+    Union,
     cast,
 )
 
 from ruamel.yaml.comments import CommentedMap
 
+from .context import LoadingContext
+from .load_tool import load_tool
 from .process import Process
 from .utils import CWLObjectType, aslist
 from .workflow import Workflow, WorkflowStep
@@ -57,16 +60,38 @@ def declare_node(nodes: Dict[str, Node], nodeid: str, tp: Optional[str]) -> Node
 
 
 def find_step(
-    steps: List[WorkflowStep], stepid: str
+    steps: List[WorkflowStep], stepid: str, loading_context: LoadingContext
 ) -> 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, index
+        if stepid.startswith(st.tool["id"]):
+            run: Union[str, Process] = st.tool["run"]
+            if isinstance(run, Workflow):
+                result, index = find_step(
+                    run.steps, stepid[len(st.tool["id"]) + 1 :], loading_context
+                )
+                if result:
+                    return result, index
+            elif isinstance(run, str):
+                process = load_tool(run, loading_context)
+                if isinstance(process, Workflow):
+                    suffix = stepid[len(st.tool["id"]) + 1 :]
+                    prefix = process.tool["id"]
+                    adj_stepid = f"{prefix}#{suffix}"
+                    result, index = find_step(
+                        process.steps, adj_stepid, loading_context
+                    )
+                    if result:
+                        return result, index
     return None, -1
 
 
-def get_subgraph(roots: MutableSequence[str], tool: Workflow) -> CommentedMap:
+def get_subgraph(
+    roots: MutableSequence[str], tool: Workflow, loading_context: LoadingContext
+) -> CommentedMap:
+    """Extract the subgraph for the given roots."""
     if tool.tool["class"] != "Workflow":
         raise Exception("Can only extract subgraph from workflow")
 
@@ -128,7 +153,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)[0]
+                        wfstep = find_step(tool.steps, v, loading_context)[0]
                         if wfstep is not None:
                             for inp in cast(
                                 MutableSequence[CWLObjectType], wfstep["inputs"]
@@ -169,11 +194,13 @@ def get_subgraph(roots: MutableSequence[str], tool: Workflow) -> CommentedMap:
     return extracted
 
 
-def get_step(tool: Workflow, step_id: str) -> CommentedMap:
-
+def get_step(
+    tool: Workflow, step_id: str, loading_context: LoadingContext
+) -> CommentedMap:
+    """Extract a single WorkflowStep for the given step_id."""
     extracted = CommentedMap()
 
-    step = find_step(tool.steps, step_id)[0]
+    step = find_step(tool.steps, step_id, loading_context)[0]
     if step is None:
         raise Exception(f"Step {step_id} was not found")
 
@@ -202,17 +229,19 @@ def get_step(tool: Workflow, step_id: str) -> CommentedMap:
 
 
 def get_process(
-    tool: Workflow, step_id: str, index: Mapping[str, Any]
+    tool: Workflow, step_id: str, loading_context: LoadingContext
 ) -> Tuple[Any, int]:
     """Find the underlying Process for a given Workflow step id."""
-    step, step_pos = find_step(tool.steps, step_id)
+    if loading_context.loader is None:
+        raise Exception("loadingContext.loader cannot be None")
+    step, step_pos = find_step(tool.steps, step_id, loading_context)
     if step is None:
         raise Exception(f"Step {step_id} was not found")
 
-    run = step["run"]
+    run: Union[str, Any] = step["run"]
 
     if isinstance(run, str):
-        process = index[run]
+        process = loading_context.loader.idx[run]
     else:
         process = run
     return process, step_pos

cwltool/workflow.py

@@ -20,7 +20,7 @@
 from schema_salad.sourceline import SourceLine, indent
 
 from . import command_line_tool, context, procgenerator
-from .checker import static_checker, circular_dependency_checker
+from .checker import circular_dependency_checker, static_checker
 from .context import LoadingContext, RuntimeContext, getdefault
 from .errors import WorkflowException
 from .load_tool import load_tool

tests/subgraph/env-tool2_no_env.cwl

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

tests/subgraph/env-tool2_req.cwl

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

tests/subgraph/env-wf2_hint_collision.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
+
+hints:
+  EnvVarRequirement:
+    envDef:
+      TEST_ENV: override
+
+steps:
+  step1:
+    run: env-tool2.cwl
+    in:
+      in: in
+    out: [out]

tests/subgraph/env-wf2_hint_req_collision.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
+
+hints:
+  EnvVarRequirement:
+    envDef:
+      TEST_ENV: override
+
+steps:
+  step1:
+    run: env-tool2_req.cwl
+    in:
+      in: in
+    out: [out]

tests/subgraph/env-wf2_only_hint.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
+
+hints:
+  EnvVarRequirement:
+    envDef:
+      TEST_ENV: override-from-parent-hint
+
+steps:
+  step1:
+    run: env-tool2_no_env.cwl
+    in:
+      in: in
+    out: [out]