Propagate "evaluate_env" workflow parameter to subworkflows

Change-Id: I6deafd8d71172faf74553a9a741753b7035725df
(cherry picked from commit deb01bb800)

mistral/engine/actions.py

@@ -472,6 +472,7 @@ class WorkflowAction(Action):
 
         if 'env' in parent_wf_ex.params:
             wf_params['env'] = parent_wf_ex.params['env']
+            wf_params['evaluate_env'] = parent_wf_ex.params.get('evaluate_env')
 
         for k, v in list(input_dict.items()):
             if k not in wf_spec.get_input():

mistral/tests/unit/engine/test_environment.py

@@ -267,3 +267,89 @@ class EnvironmentTest(base.EngineTestCase):
                 },
                 t.published
             )
+
+    def test_evaluate_env_parameter_subworkflow(self):
+        wf_text = """---
+        version: '2.0'
+
+        parent_wf:
+          tasks:
+            task1:
+              workflow: sub_wf
+
+        sub_wf:
+          output:
+            result: <% $.result %>
+
+          tasks:
+            task1:
+              action: std.noop
+              publish:
+                result: <% env().dummy %>
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        # Run with 'evaluate_env' set to False.
+
+        env = {"dummy": "<% $.ENSURE.MISTRAL.DOESNT.EVALUATE.ENV %>"}
+
+        parent_wf_ex = self.engine.start_workflow(
+            'parent_wf',
+            {},
+            env=env,
+            evaluate_env=False
+        )
+
+        self.await_workflow_success(parent_wf_ex.id)
+
+        with db_api.transaction():
+            parent_wf_ex = db_api.get_workflow_execution(parent_wf_ex.id)
+
+            t = self._assert_single_item(
+                parent_wf_ex.task_executions,
+                name='task1'
+            )
+
+            sub_wf_ex = db_api.get_workflow_executions(
+                task_execution_id=t.id
+            )[0]
+
+            self.assertDictEqual(
+                {
+                    "result": "<% $.ENSURE.MISTRAL.DOESNT.EVALUATE.ENV %>"
+                },
+                sub_wf_ex.output
+            )
+
+        # Run with 'evaluate_env' set to True.
+
+        env = {"dummy": "<% 1 + 1 %>"}
+
+        parent_wf_ex = self.engine.start_workflow(
+            'parent_wf',
+            {},
+            env=env,
+            evaluate_env=True
+        )
+
+        self.await_workflow_success(parent_wf_ex.id)
+
+        with db_api.transaction():
+            parent_wf_ex = db_api.get_workflow_execution(parent_wf_ex.id)
+
+            t = self._assert_single_item(
+                parent_wf_ex.task_executions,
+                name='task1'
+            )
+
+            sub_wf_ex = db_api.get_workflow_executions(
+                task_execution_id=t.id
+            )[0]
+
+            self.assertDictEqual(
+                {
+                    "result": 2
+                },
+                sub_wf_ex.output
+            )