Fix 'with-items' expression evaluation

* There was a bug left after the recent refactoring. While
  evaluating 'with-items' expression we didn't construct a context
  view properly, it didn't include a workflow environment. This
  patch fixes it.

Closes-Bug: #1839840
Change-Id: I3df711ef2484374418085fe0117fe8b37ce5ba3f

mistral/engine/tasks.py

@@ -825,16 +825,8 @@ class WithItemsTask(RegularTask):
 
         :return: Evaluated 'with-items' expression values.
         """
-        ctx_view = data_flow.ContextView(
-            self.ctx,
-            self.wf_ex.context,
-            self.wf_ex.input
-        )
 
-        return expr.evaluate_recursively(
-            self.task_spec.get_with_items(),
-            ctx_view
-        )
+        return self._evaluate_expression(self.task_spec.get_with_items())
 
     def _validate_values(self, with_items_values):
         # Take only mapped values and check them.

mistral/tests/unit/engine/test_with_items.py

@@ -1186,6 +1186,41 @@ class WithItemsEngineTest(base.EngineTestCase):
 
         self.assertEqual(states.SUCCESS, task1.state)
 
+    def test_with_items_env_in_with_items_expression(self):
+        wf_text = """---
+        version: "2.0"
+
+        wf:
+          tasks:
+            task1:
+              with-items: env_param in <% env().input_array %>
+              action: std.echo output=<% $.env_param %>
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow(
+            'wf',
+            env={'input_array': ['1', '2', '33']}
+        )
+
+        self.await_workflow_success(wf_ex.id, timeout=10)
+
+        with db_api.transaction():
+            # Note: We need to reread execution to access related tasks.
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            task1 = self._assert_single_item(
+                wf_ex.task_executions,
+                name='task1'
+            )
+
+            result = data_flow.get_task_execution_result(task1)
+
+        self.assertListEqual(['1', '2', '33'], result)
+        self.assertEqual(states.SUCCESS, task1.state)
+
     def test_with_items_two_tasks_second_starts_on_success(self):
         wb_text = """---
         version: "2.0"