Merge "Process input defaults and output transforms for nested AdHoc Actions"

mistral/engine/actions.py

@@ -393,14 +393,15 @@ class AdHocAction(PythonAction):
         )
 
     def _prepare_input(self, input_dict):
-        for k, v in self.action_spec.get_input().items():
-            if k not in input_dict or input_dict[k] is utils.NotDefined:
-                input_dict[k] = v
-
         base_input_dict = input_dict
 
         for action_def in self.adhoc_action_defs:
             action_spec = spec_parser.get_action_spec(action_def.spec)
+            for k, v in action_spec.get_input().items():
+                if (k not in base_input_dict or
+                        base_input_dict[k] is utils.NotDefined):
+                    base_input_dict[k] = v
+
             base_input_expr = action_spec.get_base_input()
 
             if base_input_expr:
@@ -421,15 +422,17 @@ class AdHocAction(PythonAction):
     def _prepare_output(self, result):
         # In case of error, we don't transform a result.
         if not result.is_error():
+            for action_def in reversed(self.adhoc_action_defs):
                 adhoc_action_spec = spec_parser.get_action_spec(
-                self.adhoc_action_def.spec
+                    action_def.spec
                 )
 
                 transformer = adhoc_action_spec.get_output()
 
                 if transformer is not None:
                     result = ml_actions.Result(
-                    data=expr.evaluate_recursively(transformer, result.data),
+                        data=expr.evaluate_recursively(transformer,
+                                                       result.data),
                         error=result.error
                     )
 

mistral/tests/unit/engine/test_adhoc_actions.py

@@ -46,6 +46,15 @@ actions:
     base-input:
       output: '{{ env().foo }}'
 
+  nested_concat:
+    base: my_wb.concat_twice
+    base-input:
+      s2: '{{ _.n2 }}'
+    input:
+      - n2: 'b'
+    output:
+      nested_concat: '{{ _ }}'
+
 workflows:
   wf1:
     type: direct
@@ -100,6 +109,15 @@ workflows:
         publish:
           printenv_result: '{{ task().result }}'
 
+  wf5:
+    type: direct
+    output:
+      workflow_result: '{{ _.nested_result }}'
+    tasks:
+      nested_test:
+        action: nested_concat
+        publish:
+          nested_result: '{{ task().result }}'
 """
 
 
@@ -172,3 +190,18 @@ class AdhocActionsTest(base.EngineTestCase):
                 },
                 wf_ex.output
             )
+
+    def test_run_nested_adhoc_with_output(self):
+        wf_ex = self.engine.start_workflow(
+            'my_wb.wf5', '')
+
+        self.await_workflow_success(wf_ex.id)
+        with db_api.transaction():
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            self.assertDictEqual(
+                {
+                    'workflow_result': {'nested_concat': 'a+b and a+b'}
+                },
+                wf_ex.output
+            )