16957 check circular dep (#1543)

Co-authored-by: Michael R. Crusoe <[email protected]>

Author: lijiayong

cwltool/checker.py

@@ -437,6 +437,73 @@ def check_all_types(
     return validation
 
 
+def circular_dependency_checker(step_inputs: List[CWLObjectType]) -> None:
+    """Check if a workflow has circular dependency."""
+    adjacency = get_dependency_tree(step_inputs)
+    vertices = adjacency.keys()
+    processed: List[str] = []
+    cycles: List[List[str]] = []
+    for vertex in vertices:
+        if vertex not in processed:
+            traversal_path = [vertex]
+            processDFS(adjacency, traversal_path, processed, cycles)
+    if cycles:
+        exception_msg = "The following steps have circular dependency:\n"
+        cyclestrs = [str(cycle) for cycle in cycles]
+        exception_msg += "\n".join(cyclestrs)
+        raise ValidationException(exception_msg)
+
+
+def get_dependency_tree(step_inputs: List[CWLObjectType]) -> Dict[str, List[str]]:
+    """Get the dependency tree in the form of adjacency list."""
+    adjacency = {}  # adjacency list of the dependency tree
+    for step_input in step_inputs:
+        if "source" in step_input:
+            if isinstance(step_input["source"], list):
+                vertices_in = [
+                    get_step_id(cast(str, src)) for src in step_input["source"]
+                ]
+            else:
+                vertices_in = [get_step_id(cast(str, step_input["source"]))]
+            vertex_out = get_step_id(cast(str, step_input["id"]))
+            for vertex_in in vertices_in:
+                if vertex_in not in adjacency:
+                    adjacency[vertex_in] = [vertex_out]
+                elif vertex_out not in adjacency[vertex_in]:
+                    adjacency[vertex_in].append(vertex_out)
+            if vertex_out not in adjacency:
+                adjacency[vertex_out] = []
+    return adjacency
+
+
+def processDFS(
+    adjacency: Dict[str, List[str]],
+    traversal_path: List[str],
+    processed: List[str],
+    cycles: List[List[str]],
+) -> None:
+    """Perform depth first search."""
+    tip = traversal_path[-1]
+    for vertex in adjacency[tip]:
+        if vertex in traversal_path:
+            i = traversal_path.index(vertex)
+            cycles.append(traversal_path[i:])
+        elif vertex not in processed:
+            traversal_path.append(vertex)
+            processDFS(adjacency, traversal_path, processed, cycles)
+    processed.append(tip)
+    traversal_path.pop()
+
+
+def get_step_id(field_id: str) -> str:
+    """Extract step id from either input or output fields."""
+    if "/" in field_id.split("#")[1]:
+        step_id = "/".join(field_id.split("/")[:-1])
+    else:
+        step_id = field_id.split("#")[0]
+    return step_id
+
+
 def is_conditional_step(param_to_step: Dict[str, CWLObjectType], parm_id: str) -> bool:
     source_step = param_to_step.get(parm_id)
     if source_step is not None:

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
+from .checker import static_checker, circular_dependency_checker
 from .context import LoadingContext, RuntimeContext, getdefault
 from .errors import WorkflowException
 from .load_tool import load_tool
@@ -139,6 +139,7 @@ def __init__(
                 step_outputs,
                 param_to_step,
             )
+            circular_dependency_checker(step_inputs)
 
     def make_workflow_step(
         self,

tests/checker_wf/cat-a.cwl

@@ -0,0 +1,12 @@
+#!/usr/bin/env cwl-runner
+cwlVersion: v1.1
+class: CommandLineTool
+inputs:
+  intxt:
+    type: File[]
+    inputBinding: {}
+outputs:
+  cattxt:
+    type: stdout
+baseCommand: [cat, -A]
+stdout: cat.txt

tests/checker_wf/circ-dep-wf.cwl

@@ -0,0 +1,36 @@
+#!/usr/bin/env cwl-runner
+cwlVersion: v1.1
+class: Workflow
+requirements:
+  MultipleInputFeatureRequirement: {}
+
+inputs:
+  txt:
+    type: File
+    default:
+      class: File
+      location: test.txt
+
+outputs:
+  wctxt:
+    type: File
+    outputSource: wc/wctxt
+
+steps:
+  cat-a:
+    run: cat-a.cwl
+    in:
+      intxt:
+        source: [txt, wc/wctxt]
+        linkMerge: merge_flattened
+    out: [cattxt]
+  ls:
+    run: ls.cwl
+    in:
+      intxt: cat-a/cattxt
+    out: [lstxt]
+  wc:
+    run: wc.cwl
+    in:
+      intxt: ls/lstxt
+    out: [wctxt]

tests/checker_wf/circ-dep-wf2.cwl

@@ -0,0 +1,36 @@
+#!/usr/bin/env cwl-runner
+cwlVersion: v1.1
+class: Workflow
+requirements:
+  MultipleInputFeatureRequirement: {}
+
+inputs:
+  txt:
+    type: File
+    default:
+      class: File
+      location: test.txt
+
+outputs:
+  wctxt:
+    type: File
+    outputSource: wc/wctxt
+
+steps:
+  cat-a:
+    run: cat-a.cwl
+    in:
+      intxt:
+        source: [txt, txt]
+        linkMerge: merge_flattened
+    out: [cattxt]
+  ls:
+    run: ls.cwl
+    in:
+      intxt: ls/lstxt
+    out: [lstxt]
+  wc:
+    run: wc.cwl
+    in:
+      intxt: ls/lstxt
+    out: [wctxt]

tests/checker_wf/ls.cwl

@@ -0,0 +1,12 @@
+#!/usr/bin/env cwl-runner
+cwlVersion: v1.1
+class: CommandLineTool
+inputs:
+  intxt:
+    type: File
+    inputBinding: {}
+outputs:
+  lstxt:
+    type: stdout
+baseCommand: ls
+stdout: ls.txt

tests/checker_wf/test.txt

@@ -0,0 +1 @@
+a

tests/checker_wf/wc.cwl

@@ -0,0 +1,12 @@
+#!/usr/bin/env cwl-runner
+cwlVersion: v1.1
+class: CommandLineTool
+inputs:
+  intxt:
+    type: File
+    inputBinding: {}
+outputs:
+  wctxt:
+    type: stdout
+baseCommand: wc
+stdout: wc.txt

tests/test_examples.py

@@ -921,6 +921,7 @@ def test_format_expr_error() -> None:
 def test_static_checker() -> None:
     # check that the static checker raises exception when a source type
     # mismatches its sink type.
+    """Confirm that static type checker raises expected exception."""
     factory = cwltool.factory.Factory()
 
     with pytest.raises(ValidationException):
@@ -933,6 +934,49 @@ def test_static_checker() -> None:
         factory.make(get_data("tests/checker_wf/broken-wf3.cwl"))
 
 
+def test_circular_dependency_checker() -> None:
+    # check that the circular dependency checker raises exception when there is
+    # circular dependency in the workflow.
+    """Confirm that circular dependency checker raises expected exception."""
+    factory = cwltool.factory.Factory()
+
+    with pytest.raises(
+        ValidationException,
+        match=r".*The\s*following\s*steps\s*have\s*circular\s*dependency:\s*.*",
+    ):
+        factory.make(get_data("tests/checker_wf/circ-dep-wf.cwl"))
+
+    with pytest.raises(
+        ValidationException,
+        match=r".*#cat-a.*",
+    ):
+        factory.make(get_data("tests/checker_wf/circ-dep-wf.cwl"))
+
+    with pytest.raises(
+        ValidationException,
+        match=r".*#ls.*",
+    ):
+        factory.make(get_data("tests/checker_wf/circ-dep-wf.cwl"))
+
+    with pytest.raises(
+        ValidationException,
+        match=r".*#wc.*",
+    ):
+        factory.make(get_data("tests/checker_wf/circ-dep-wf.cwl"))
+
+    with pytest.raises(
+        ValidationException,
+        match=r".*The\s*following\s*steps\s*have\s*circular\s*dependency:\s*.*",
+    ):
+        factory.make(get_data("tests/checker_wf/circ-dep-wf2.cwl"))
+
+    with pytest.raises(
+        ValidationException,
+        match=r".*#ls.*",
+    ):
+        factory.make(get_data("tests/checker_wf/circ-dep-wf2.cwl"))
+
+
 def test_var_spool_cwl_checker1() -> None:
     """Confirm that references to /var/spool/cwl are caught."""
     stream = StringIO()