Dynamic action name evaluation.

Evaluate action names dynamically, so yaql or jinja expression can be
used

Change-Id: I48761c215f0255976c330ffa34f27bb695c944a9
Implements: blueprint mistral-dynamic-actions

doc/source/user/dsl_v2.rst

@@ -188,7 +188,8 @@ attributes:
 
 -  **description** - Arbitrary text containing task description.
    *Optional*.
--  **action** - Name of the action associated with the task.
+-  **action** - Name of the action associated with the task.Can be static
+   value or an expression (for example,  "{{ _.action_name }}").
    *Mutually exclusive with* **workflow**. If neither action nor workflow are
    provided then the action 'std.noop' will be used.
 -  **workflow** - Name of the workflow associated with the task. Can be static

mistral/engine/tasks.py

@@ -457,12 +457,17 @@ class RegularTask(Task):
         action_name = self.task_spec.get_action_name()
         wf_name = self.task_spec.get_workflow_name()
 
+        # For dynamic workflow evaluation we regenerate the action.
         if wf_name:
             return actions.WorkflowAction(
                 wf_name=self._evaluate_expression(wf_name),
                 task_ex=self.task_ex
             )
 
+        # For dynamic action evaluation we just regenerate the name.
+        if action_name:
+            action_name = self._evaluate_expression(action_name)
+
         if not action_name:
             action_name = 'std.noop'
 

mistral/lang/base.py

@@ -26,13 +26,13 @@ from mistral.expressions.yaql_expression import INLINE_YAQL_REGEXP
 from mistral.lang import types
 from mistral import utils
 
-ACTION_PATTRENS = {
+ACTION_PATTERNS = {
     "command": "[\w\.]+[^=\(\s\"]*",
     "yaql_expression": INLINE_YAQL_REGEXP,
     "jinja_expression": ANY_JINJA_REGEXP,
 }
 CMD_PTRN = re.compile(
-    "^({})".format("|".join(six.itervalues(ACTION_PATTRENS)))
+    "^({})".format("|".join(six.itervalues(ACTION_PATTERNS)))
 )
 
 EXPRESSION = '|'.join([expr.patterns[name] for name in expr.patterns])

mistral/tests/unit/engine/test_workflow_variables.py

@@ -73,3 +73,84 @@ class WorkflowVariablesTest(base.EngineTestCase):
             },
             wf_output
         )
+
+    def test_dynamic_action_names(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf2:
+          input:
+            - wf_action
+            - param1
+
+          tasks:
+            task1:
+              action: <% $.wf_action %> output=<% $.param1 %>
+              publish:
+                var1: <% task(task1).result %>
+
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow(
+            'wf2',
+            "",
+            {"wf_action": "std.echo", "param1": "Hello"}
+        )
+
+        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)
+
+            wf_output = wf_ex.output
+            tasks = wf_ex.task_executions
+
+        task1 = self._assert_single_item(tasks, name='task1')
+
+        self.assertEqual(states.SUCCESS, task1.state)
+        self.assertEqual("Hello", wf_output['var1'])
+
+    def test_dynamic_action_names_and_input(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf3:
+          input:
+            - wf_action
+            - wf_input
+
+          tasks:
+            task1:
+              action: <% $.wf_action %>
+              input: <% $.wf_input %>
+              publish:
+                var1: <% task(task1).result %>
+
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow(
+            'wf3',
+            "",
+            {"wf_action": "std.echo", "wf_input": {"output": "Hello"}}
+        )
+
+        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)
+
+            wf_output = wf_ex.output
+            tasks = wf_ex.task_executions
+
+        task1 = self._assert_single_item(tasks, name='task1')
+
+        self.assertEqual(states.SUCCESS, task1.state)
+        self.assertEqual("Hello", wf_output['var1'])