Add YAQL function 'task' to extract info about task in DSL

Change-Id: I370e7f1ca751ca278b8142304e3f5fcf83613c20

mistral/tests/unit/engine/test_yaql_functions.py

@@ -0,0 +1,111 @@
+# Copyright 2015 - Mirantis, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+
+from oslo_config import cfg
+from oslo_log import log as logging
+
+from mistral.db.v2 import api as db_api
+from mistral.services import workflows as wf_service
+from mistral.tests.unit.engine import base as engine_test_base
+from mistral.workflow import states
+
+LOG = logging.getLogger(__name__)
+
+# Use the set_default method to set value otherwise in certain test cases
+# the change in value is not permanent.
+cfg.CONF.set_default('auth_enable', False, group='pecan')
+
+
+class YAQLFunctionsEngineTest(engine_test_base.EngineTestCase):
+    def test_task_function(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          type: direct
+
+          tasks:
+            task1:
+              description: This is task 1
+              tags: ['t1']
+              action: std.echo output=1
+              publish:
+                name: <% task(task1).name %>
+                description: <% task(task1).spec.description %>
+                tags: <% task(task1).spec.tags%>
+                state: <% task(task1).state %>
+                state_info: <% task(task1).state_info %>
+                res: <% task(task1).result %>
+              on-success:
+                - task2
+
+            task2:
+              action: std.echo output=<% task(task1).result + 1 %>
+              publish:
+                name: <% task(task1).name %>
+                description: <% task(task1).spec.description %>
+                tags: <% task(task1).spec.tags%>
+                state: <% task(task1).state %>
+                state_info: <% task(task1).state_info %>
+                res: <% task(task1).result %>
+                task2_res: <% task(task2).result %>
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        wf_ex = self.engine.start_workflow('wf', {})
+
+        self._await(lambda: self.is_execution_success(wf_ex.id), timeout=5)
+
+        # Reread execution to access related tasks.
+        wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+        self.assertEqual(states.SUCCESS, wf_ex.state)
+
+        tasks = wf_ex.task_executions
+
+        task1 = self._assert_single_item(
+            tasks,
+            name='task1',
+            state=states.SUCCESS
+        )
+        task2 = self._assert_single_item(
+            tasks,
+            name='task2',
+            state=states.SUCCESS
+        )
+
+        self.assertDictEqual(
+            {
+                'name': 'task1',
+                'description': 'This is task 1',
+                'tags': ['t1'],
+                'state': states.SUCCESS,
+                'state_info': None,
+                'res': 1
+            },
+            task1.published
+        )
+        self.assertDictEqual(
+            {
+                'name': 'task1',
+                'description': 'This is task 1',
+                'tags': ['t1'],
+                'state': states.SUCCESS,
+                'state_info': None,
+                'res': 1,
+                'task2_res': 2
+            },
+            task2.published
+        )

mistral/utils/yaql_utils.py

@@ -14,6 +14,8 @@
 
 import yaql
 
+from mistral.db.v2 import api as db_api
+from mistral.workflow import utils as wf_utils
 
 ROOT_CONTEXT = None
 
@@ -23,6 +25,7 @@ def get_yaql_context(data_context):
 
     if not ROOT_CONTEXT:
         ROOT_CONTEXT = yaql.create_context()
+
         _register_functions(ROOT_CONTEXT)
 
     new_ctx = ROOT_CONTEXT.create_child_context()
@@ -38,9 +41,10 @@ def get_yaql_context(data_context):
 def _register_functions(yaql_ctx):
     yaql_ctx.register_function(env_)
     yaql_ctx.register_function(execution_)
+    yaql_ctx.register_function(task_)
 
 
-# Additional convenience YAQL functions.
+# Additional YAQL functions needed by Mistral.
 # If a function name ends with underscore then it doesn't need to pass
 # the name of the function when context registers it.
 
@@ -50,3 +54,33 @@ def env_(context):
 
 def execution_(context):
     return context['__execution']
+
+
+def task_(context, task_name):
+    # Importing data_flow in order to break cycle dependency between modules.
+    from mistral.workflow import data_flow
+
+    wf_ex = db_api.get_workflow_execution(context['__execution']['id'])
+
+    task_execs = wf_utils.find_task_executions_by_name(wf_ex, task_name)
+
+    # TODO(rakhmerov): Account for multiple executions (i.e. in case of
+    # cycles).
+    task_ex = task_execs[-1]
+
+    if not task_ex:
+        raise ValueError(
+            'Failed to find task execution with name: %s' % task_name
+        )
+
+    # We don't use to_dict() db model method because not all fields
+    # make sense for user.
+    return {
+        'id': task_ex.id,
+        'name': task_ex.name,
+        'spec': task_ex.spec,
+        'state': task_ex.state,
+        'state_info': task_ex.state_info,
+        'result': data_flow.get_task_execution_result(task_ex),
+        'published': task_ex.published
+    }

mistral/workflow/utils.py

@@ -63,11 +63,12 @@ def find_task_execution_with_state(wf_ex, task_spec, state):
     return task_execs[0] if len(task_execs) > 0 else None
 
 
+def find_task_executions_by_name(wf_ex, task_name):
+    return [t for t in wf_ex.task_executions if t.name == task_name]
+
+
 def find_task_executions_by_spec(wf_ex, task_spec):
-    return [
+    return find_task_executions_by_name(wf_ex, task_spec.get_name())
-        t for t in wf_ex.task_executions
-        if t.name == task_spec.get_name()
-    ]
 
 
 def find_task_executions_by_specs(wf_ex, task_specs):