Add 'evaluate_env' workflow parameter

* This parameter, if set to False, prevents from evaluating
  expressions in a workflow environment. By default, it's set to
  True for backwards compatibility. In some cases, it guarantees
  safety from blowing size of the evaluated objects (e.g. if
  there are expressions in it like <% $ %> and <% env() %>). Or
  if we simply don't want to evaluate it because we want to have
  a text in the environment containing expressions as informative.

Change-Id: I116004e9e2d0269fc69e39b552165f6e0720148a
Closes-Bug: #1690319

mistral/tests/unit/engine/test_environment.py

@@ -19,6 +19,7 @@ from mistral.db.v2 import api as db_api
 from mistral.executors import default_executor as d_exe
 from mistral.executors import remote_executor as r_exe
 from mistral.services import workbooks as wb_service
+from mistral.services import workflows as wf_service
 from mistral.tests.unit.engine import base
 
 
@@ -199,3 +200,71 @@ class EnvironmentTest(base.EngineTestCase):
         }
 
         self._test_subworkflow(env)
+
+    def test_evaluate_env_parameter(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          tasks:
+            task1:
+              action: std.noop
+              publish:
+                var1: <% env().var1 %>
+                var2: <% env().var2 %>
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        env = {
+            "var1": "val1",
+            "var2": "<% env().var1 %>"
+        }
+
+        # Run with 'evaluate_env' set to True.
+
+        wf_ex = self.engine.start_workflow(
+            'wf',
+            {},
+            env=env,
+            evaluate_env=True
+        )
+
+        self.await_workflow_success(wf_ex.id)
+
+        with db_api.transaction():
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            t = self._assert_single_item(wf_ex.task_executions, name='task1')
+
+            self.assertDictEqual(
+                {
+                    "var1": "val1",
+                    "var2": "val1"
+                },
+                t.published
+            )
+
+        # Run with 'evaluate_env' set to False.
+
+        wf_ex = self.engine.start_workflow(
+            'wf',
+            {},
+            env=env,
+            evaluate_env=False
+        )
+
+        self.await_workflow_success(wf_ex.id)
+
+        with db_api.transaction():
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            t = self._assert_single_item(wf_ex.task_executions, name='task1')
+
+            self.assertDictEqual(
+                {
+                    "var1": "val1",
+                    "var2": "<% env().var1 %>"
+                },
+                t.published
+            )

mistral/workflow/data_flow.py

@@ -297,8 +297,14 @@ def add_environment_to_context(wf_ex):
     if 'env' in wf_ex.params:
         env = copy.deepcopy(wf_ex.params['env'])
 
-        # An env variable can be an expression of other env variables.
-        wf_ex.context['__env'] = expr.evaluate_recursively(env, {'__env': env})
+        if ('evaluate_env' in wf_ex.params and
+                not wf_ex.params['evaluate_env']):
+            wf_ex.context['__env'] = env
+        else:
+            wf_ex.context['__env'] = expr.evaluate_recursively(
+                env,
+                {'__env': env}
+            )
 
 
 def add_workflow_variables_to_context(wf_ex, wf_spec):