Merge "Advanced publishing: add 'global' function to access global variables"

mistral/tests/unit/engine/test_dataflow.py

@@ -832,6 +832,100 @@ class DataFlowEngineTest(engine_test_base.EngineTestCase):
 
         self.assertDictEqual({'result': 'We got an error'}, task2.published)
 
+    def test_global_publishing_success_access_via_function(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          tasks:
+            task1:
+              action: std.noop
+              on-success:
+                publish:
+                  branch:
+                    my_var: Branch local value
+                  global:
+                    my_var: Global value
+                next:
+                  - task2
+
+            task2:
+              action: std.noop
+              publish:
+                local: <% $.my_var %>
+                global: <% global(my_var) %>
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        wf_ex = self.engine.start_workflow('wf', {})
+
+        self.await_workflow_success(wf_ex.id)
+
+        with db_api.transaction():
+            # Note: We need to reread execution to access related tasks.
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            tasks = wf_ex.task_executions
+
+        self._assert_single_item(tasks, name='task1')
+        task2 = self._assert_single_item(tasks, name='task2')
+
+        self.assertDictEqual(
+            {
+                'local': 'Branch local value',
+                'global': 'Global value'
+            },
+            task2.published
+        )
+
+    def test_global_publishing_error_access_via_function(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          tasks:
+            task1:
+              action: std.fail
+              on-error:
+                publish:
+                  branch:
+                    my_var: Branch local value
+                  global:
+                    my_var: Global value
+                next:
+                  - task2
+
+            task2:
+              action: std.noop
+              publish:
+                local: <% $.my_var %>
+                global: <% global(my_var) %>
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        wf_ex = self.engine.start_workflow('wf', {})
+
+        self.await_workflow_success(wf_ex.id)
+
+        with db_api.transaction():
+            # Note: We need to reread execution to access related tasks.
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            tasks = wf_ex.task_executions
+
+        self._assert_single_item(tasks, name='task1')
+        task2 = self._assert_single_item(tasks, name='task2')
+
+        self.assertDictEqual(
+            {
+                'local': 'Branch local value',
+                'global': 'Global value'
+            },
+            task2.published
+        )
+
 
 class DataFlowTest(test_base.BaseTest):
     def test_get_task_execution_result(self):

mistral/utils/expression_utils.py

@@ -262,3 +262,9 @@ def _convert_to_user_model(task_ex):
 
 def uuid_(context=None):
     return utils.generate_unicode_uuid()
+
+
+def global_(context, var_name):
+    wf_ex = db_api.get_workflow_execution(context['__execution']['id'])
+
+    return wf_ex.context.get(var_name)

setup.cfg

@@ -71,6 +71,7 @@ mistral.actions =
     std.test_dict = mistral.actions.std_actions:TestDictAction
 
 mistral.expression.functions =
+    global = mistral.utils.expression_utils:global_
     json_pp = mistral.utils.expression_utils:json_pp_
     task = mistral.utils.expression_utils:task_
     tasks = mistral.utils.expression_utils:tasks_