Result will be [], if list for with-items is empty.

Change-Id: Ia05c5c318c7a5b8863f45e1955ed346e80f0e913
Closes-Bug: #1512477

mistral/tests/unit/engine/test_dataflow.py

@@ -461,6 +461,36 @@ class DataFlowEngineTest(engine_test_base.EngineTestCase):
         # But all result is cleared.
         self.assertIsNone(result)
 
+    def test_empty_with_items(self):
+        wf = """---
+        version: "2.0"
+
+        wf1_with_items:
+           type: direct
+
+           tasks:
+             task1:
+               with-items: i in <% list() %>
+               action: std.echo output= "Task 1.<% $.i %>"
+               publish:
+                 result: <% $.task1 %>
+        """
+        wf_service.create_workflows(wf)
+
+        # Start workflow.
+        wf_ex = self.engine.start_workflow('wf1_with_items', {})
+
+        self._await(lambda: self.is_execution_success(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.assertListEqual([], result)
+
 
 class DataFlowTest(test_base.BaseTest):
     def test_get_task_execution_result(self):
@@ -471,6 +501,9 @@ class DataFlowTest(test_base.BaseTest):
                 'name': 'task1',
                 'with-items': 'var in [1]',
                 'type': 'direct'
+            },
+            runtime_context={
+                'with_items_context': {'count': 1}
             }
         )
 

mistral/workflow/data_flow.py

@@ -27,7 +27,7 @@ from mistral import utils
 from mistral.utils import inspect_utils
 from mistral.workbook import parser as spec_parser
 from mistral.workflow import states
-
+from mistral.workflow import with_items
 
 LOG = logging.getLogger(__name__)
 CONF = cfg.CONF
@@ -107,7 +107,10 @@ def get_task_execution_result(task_ex):
     task_spec = spec_parser.get_task_spec(task_ex.spec)
 
     if task_spec.get_with_items():
+        if with_items.get_count(task_ex) > 0:
             return results
+        else:
+            return []
 
     return results[0] if len(results) == 1 else results