❤️️ TEST: Updated

Author: FernandoCelmer

tests/test_flow.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python
+
+from dotflow import DotFlow, action
+
+
+@action(retry=5)
+class StepX:
+
+    @action
+    def first_function(self):
+        return {"foo": "bar"}
+
+    @action
+    def second_function(self):
+        return True
+
+
+@action(retry=5)
+class StepY:
+
+    def __init__(self, initial_context):
+        print(initial_context.storage, "__init__")
+        assert initial_context.storage
+        self.variable = True
+
+    def auxiliary_function(self):
+        """This function will not be executed, because
+        it does not have an 'action' decorator.
+        """
+
+    @action
+    def first_function(self, initial_context):
+        print(initial_context.storage, "first_function")
+        assert initial_context.storage
+        assert self.variable
+
+        return {"foo": "bar"}
+
+    @action
+    def second_function(self, previous_context):
+        print(previous_context.storage, "second_function")
+        assert previous_context.storage
+
+        return True
+
+
+@action
+def simple_step():
+    return "ok"
+
+
+@action
+def extract_task_x():
+    return "extract"
+
+
+@action
+def transform_task_x(initial_context, previous_context):
+    print(initial_context.storage, "initial_context")
+    print(previous_context.storage, "previous_context")
+
+    assert initial_context.storage, {"foo": True}
+    assert previous_context.storage, "extract"
+
+    return "transform"
+
+
+@action
+def load_task_x():
+    return "load"
+
+
+@action
+def extract_task_y():
+    print("extract")
+    return "extract"
+
+
+@action
+def transform_task_y(previous_context):
+    print(previous_context.storage, "transform")
+    assert previous_context.storage, "extract"
+
+    return "transform"
+
+
+@action
+def load_task_y(previous_context):
+    print(previous_context.storage, "load")
+    assert previous_context.storage, "transform"
+
+    return "load"
+
+
+def main():
+    workflow = DotFlow()
+
+    workflow.task.add(step=StepX)
+    workflow.task.add(step=StepY, initial_context={"foo": "bar"})
+    workflow.task.add(step=simple_step)
+    workflow.task.add(step=simple_step)
+    workflow.task.add(step=simple_step)
+    workflow.task.add(step=simple_step)
+    workflow.task.add(step=simple_step)
+    workflow.task.add(step=simple_step)
+    workflow.task.add(step=simple_step)
+    workflow.task.add(step=simple_step)
+    workflow.task.add(step=extract_task_x)
+    workflow.task.add(step=transform_task_x, initial_context={"foo": True})
+    workflow.task.add(step=load_task_x)
+    workflow.task.add(step=extract_task_y)
+    workflow.task.add(step=transform_task_y)
+    workflow.task.add(step=load_task_y)
+    workflow.start()
+
+    return workflow
+
+
+if __name__ == "__main__":
+    main()

tests/test_integration.py

@@ -7,7 +7,7 @@
 from examples.cli_with_mode import main as cli_with_mode
 from examples.cli_with_output_context import main as cli_with_output_context
 from examples.cli_with_path import main as cli_with_path
-from examples.cli import main as cli
+from examples.simple_cli import main as simple_cli
 from examples.simple_class_workflow import main as simple_class_workflow
 from examples.simple_function_workflow_with_error import main as simple_function_workflow_with_error
 from examples.simple_function_workflow import main as simple_function_workflow
@@ -44,8 +44,8 @@ def test_cli_with_output_context(self):
     def test_cli_with_path(self):
         cli_with_path()
 
-    def test_cli(self):
-        cli()
+    def test_simple_cli(self):
+        simple_cli()
 
     def test_simple_class_workflow(self):
         simple_class_workflow()