Merge pull request #50 from soasme/predecessors

Bugfix: set correct arrow direction for graph visualization.

ROADMAP.md

@@ -62,7 +62,7 @@
 - [ ] core: support task argument `_timeout`.
 - [ ] core: preview all task execution statuses in ascii chars.
 - [ ] core: get all task execution statuses.
-- [ ] bugfix: visualize: the arrow in the output is reversed.
+- [x] bugfix: visualize: the arrow in the output is reversed. v0.8.1.
 - [x] core: remove context from Task.run signature. v0.8.1.
 - [x] bugfix: resolve deps failed when implicit deps is in inner map. v0.8.1.
 - [x] FEATURES FOR ALPHA RELEASE

examples/sqlite3_exec.hcl

@@ -40,5 +40,9 @@ flow "sqlite3_exec" {
         value = "v3"
       }
     }
+
+    _depends_on = [
+      task.sql_exec.create_table,
+    ]
   }
 }

runflow/core.py

@@ -188,7 +188,7 @@ def __init__(self, flow):
         self.results = {}
 
     def check_depends_on(self, task):
-        for upstream_task in self.flow.graph.successors(task):
+        for upstream_task in self.flow.graph.predecessors(task):
             if self.results[upstream_task].status in (
                 TaskStatus.FAILED,
                 TaskStatus.CANCELED,
@@ -230,7 +230,7 @@ def __init__(self, name, runner_cls=None):
     def __iter__(self):
         """Iterate through all tasks in a dependent order."""
         try:
-            return reversed(list(networkx.topological_sort(self.graph)))
+            yield from networkx.topological_sort(self.graph)
         except networkx.exception.NetworkXUnfeasible as err:
             raise RunflowAcyclicTasksError(str(err)) from err
 
@@ -274,8 +274,12 @@ def add_task(self, task):
         self.graph.add_node(task)
 
     def set_dependency(self, task, depends_on):
-        """Let `task` depends on `depends_on`."""
-        self.graph.add_edge(task, depends_on)
+        """Let `task` depends on `depends_on`,
+
+        e.g. `depends_on` is a predecessor edge of `task`.
+
+        """
+        self.graph.add_edge(depends_on, task)
 
     def set_default_var(self, name, value):
         """Set default value for variable."""

tests/test_cli.py

@@ -80,3 +80,13 @@ def test_visualize(tmpdir):
     svg = output.read()
     assert "task.bash_run.echo" in svg
     assert "</svg>" in svg
+
+def test_visualize_dot(tmpdir):
+    output = tmpdir / "visualize.dot"
+    cli(['visualize', '--output', str(output), 'examples/hello-deps.hcl'])
+    assert '"task.bash_run.greeter" -> "task.bash_run.echo"' in output.read()
+
+def test_visualize_dot2(tmpdir):
+    output = tmpdir / "visualize.dot"
+    cli(['visualize', '--output', str(output), 'examples/hello-implicit-deps.hcl'])
+    assert '"task.bash_run.greeter" -> "task.bash_run.echo"' in output.read()