Fix get task list on YAQL error in with-items

If there is a YAQL error in with-items, the with_items_context is not
initialized and get task list from the API will failed with
"ERROR (app) 'with_items_context'". This patch fixes the data_flow
module and returns a default with_items_context dictionary if it
does not exist.

Change-Id: I2cc7d71f30dfad3122e06637a7583333cb02de6b

mistral/tests/unit/engine/test_with_items.py

@@ -244,6 +244,37 @@ class WithItemsEngineTest(base.EngineTestCase):
 
         self.assertEqual(2, len(wf_ex.task_executions))
 
+    def test_with_items_yaql_fail(self):
+        wf_text = """---
+        version: "2.0"
+
+        with_items:
+          type: direct
+
+          tasks:
+            task1:
+              with-items: i in <% $.foobar %>
+              action: std.noop
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow('with_items', {})
+
+        self.await_execution_error(wf_ex.id)
+
+        # Note: We need to reread execution to access related tasks.
+        wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+        tasks = wf_ex.task_executions
+        task1 = self._assert_single_item(tasks, name='task1')
+        result = data_flow.get_task_execution_result(task1)
+
+        self.assertEqual(states.ERROR, task1.state)
+        self.assertIsInstance(result, list)
+        self.assertListEqual(result, [])
+
     def test_with_items_sub_workflow_fail(self):
         wb_text = """---
         version: "2.0"

mistral/workflow/with_items.py

@@ -25,9 +25,15 @@ _CONCURRENCY = 'concurrency'
 _COUNT = 'count'
 _WITH_ITEMS = 'with_items_context'
 
+_DEFAULT_WITH_ITEMS = {
+    _COUNT: 0,
+    _CONCURRENCY: 0,
+    _CAPACITY: 0
+}
+
 
 def _get_context(task_ex):
-    return task_ex.runtime_context[_WITH_ITEMS]
+    return task_ex.runtime_context.get(_WITH_ITEMS, _DEFAULT_WITH_ITEMS)
 
 
 def get_count(task_ex):