def test_get_task_execution_result(self):
        task_ex = models.TaskExecution(name='task1')

        task_ex.executions.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=True,
            runtime_context={'with_items_index': 0}
        ))

        self.assertEqual(1, data_flow.get_task_execution_result(task_ex))

        task_ex.executions.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=True,
            runtime_context={'with_items_index': 0}
        ))
        task_ex.executions.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=False,
            runtime_context={'with_items_index': 0}
        ))

        self.assertEqual([1, 1], data_flow.get_task_execution_result(task_ex))

    def test_get_task_execution_result(self):
        task_ex = models.TaskExecution(
            name='task1',
            spec={
                "version": '2.0',
                'name': 'task1',
                'with-items': 'var in [1]',
                'type': 'direct'
            }
        )

        task_ex.executions.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=True,
            runtime_context={'with_items_index': 0}
        ))

        self.assertEqual([1], data_flow.get_task_execution_result(task_ex))

        task_ex.executions.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=True,
            runtime_context={'with_items_index': 0}
        ))
        task_ex.executions.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=False,
            runtime_context={'with_items_index': 0}
        ))

        self.assertEqual([1, 1], data_flow.get_task_execution_result(task_ex))

    def test_with_items_two_tasks_second_starts_on_success(self):
        wb_text = """---
        version: "2.0"

        name: wb1

        workflows:
          with_items:
            type: direct

            tasks:
              task1:
                with-items: i in [1, 2]
                action: std.echo output=<% $.i %>
                on-success: task2
              task2:
                with-items: i in [3, 4]
                action: std.echo output=<% $.i %>
        """

        wb_service.create_workbook_v2(wb_text)

        # Start workflow.
        wf_ex = self.engine.start_workflow('wb1.with_items', {})

        self.await_workflow_success(wf_ex.id)

        with db_api.transaction():
            # Note: We need to reread execution to access related tasks.
            wf_ex = db_api.get_workflow_execution(wf_ex.id)

            task_execs = wf_ex.task_executions

        task1_ex = self._assert_single_item(
            task_execs,
            name='task1',
            state=states.SUCCESS
        )
        task2_ex = self._assert_single_item(
            task_execs,
            name='task2',
            state=states.SUCCESS
        )

        with db_api.transaction():
            task1_ex = db_api.get_task_execution(task1_ex.id)
            task2_ex = db_api.get_task_execution(task2_ex.id)

            result_task1 = data_flow.get_task_execution_result(task1_ex)
            result_task2 = data_flow.get_task_execution_result(task2_ex)

        # Since we know that we can receive results in random order,
        # check is not depend on order of items.
        self.assertIn(1, result_task1)
        self.assertIn(2, result_task1)
        self.assertIn(3, result_task2)
        self.assertIn(4, result_task2)

    def test_error_result1(self):
        wf_service.create_workflows(WF)

        # Start workflow.
        wf_ex = self.engine.start_workflow("wf", {"success_result": None, "error_result": 2})

        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

        self.assertEqual(2, len(tasks))

        task1 = self._assert_single_item(tasks, name="task1")
        task2 = self._assert_single_item(tasks, name="task2")

        self.assertEqual(states.ERROR, task1.state)
        self.assertEqual(states.SUCCESS, task2.state)

        # "publish" clause is ignored in case of ERROR so task execution field
        # must be empty.
        self.assertDictEqual({}, task1.published)
        self.assertEqual(2, data_flow.get_task_execution_result(task1))

    def test_success_result(self):
        wf_service.create_workflows(WF)

        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'wf',
            {
                'success_result': 'success',
                'error_result': None
            }
        )

        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

        self.assertEqual(1, len(tasks))

        task1 = self._assert_single_item(tasks, name='task1')

        self.assertEqual(states.SUCCESS, task1.state)

        # "publish" clause is ignored in case of ERROR so task execution field
        # must be empty.
        self.assertDictEqual({'p_var': 'success'}, task1.published)
        self.assertEqual('success', data_flow.get_task_execution_result(task1))

    def test_with_items_results_one_item_as_list(self):
        wb_service.create_workbook_v2(WB)

        # Start workflow.
        wf_ex = self.engine.start_workflow('wb1.with_items', WF_INPUT_ONE_ITEM)

        self.await_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)

        task_execs = wf_ex.task_executions

        self.assertEqual(1, len(task_execs))

        task1_ex = self._assert_single_item(
            task_execs,
            name='task1',
            state=states.SUCCESS
        )

        result = data_flow.get_task_execution_result(task1_ex)

        self.assertIsInstance(result, list)
        self.assertIn('Guy', result)

        self.assertIn(task1_ex.published['result'], ['Guy'])

    def test_with_items_action_context(self):
        wb_service.create_workbook_v2(WORKBOOK_ACTION_CONTEXT)

        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'wb1.wf1_with_items', WF_INPUT_URLS
        )

        wf_ex = db_api.get_workflow_execution(wf_ex.id)
        task_ex = wf_ex.task_executions[0]

        act_exs = task_ex.executions
        self.engine.on_action_complete(act_exs[0].id, wf_utils.Result("Ivan"))
        self.engine.on_action_complete(act_exs[1].id, wf_utils.Result("John"))
        self.engine.on_action_complete(
            act_exs[2].id, wf_utils.Result("Mistral")
        )

        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)

        task_ex = db_api.get_task_execution(task_ex.id)
        result = data_flow.get_task_execution_result(task_ex)

        self.assertTrue(isinstance(result, list))

        self.assertIn('John', result)
        self.assertIn('Ivan', result)
        self.assertIn('Mistral', result)

        self.assertEqual(states.SUCCESS, task_ex.state)

    def test_error_result1(self):
        wf_service.create_workflows(WF)

        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'wf',
            {
                'success_result': None,
                'error_result': 2
            }
        )

        self.await_workflow_success(wf_ex.id)

        with db_api.transaction():
            # 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

            self.assertEqual(2, len(tasks))

            task1 = self._assert_single_item(tasks, name='task1')
            task2 = self._assert_single_item(tasks, name='task2')

            self.assertEqual(states.ERROR, task1.state)
            self.assertEqual(states.SUCCESS, task2.state)

            # "publish" clause is ignored in case of ERROR so task execution
            # field must be empty.
            self.assertDictEqual({}, task1.published)
            self.assertEqual(2, data_flow.get_task_execution_result(task1))

def task_(context, task_name):
    # Importing data_flow in order to break cycle dependency between modules.
    from mistral.workflow import data_flow

    wf_ex = db_api.get_workflow_execution(context['__execution']['id'])

    task_execs = wf_utils.find_task_executions_by_name(wf_ex, task_name)

    # TODO(rakhmerov): Account for multiple executions (i.e. in case of
    # cycles).
    task_ex = task_execs[-1]

    if not task_ex:
        raise ValueError(
            'Failed to find task execution with name: %s' % task_name
        )

    # We don't use to_dict() db model method because not all fields
    # make sense for user.
    return {
        'id': task_ex.id,
        'name': task_ex.name,
        'spec': task_ex.spec,
        'state': task_ex.state,
        'state_info': task_ex.state_info,
        'result': data_flow.get_task_execution_result(task_ex),
        'published': task_ex.published
    }

    def test_with_items_simple(self):
        wb_service.create_workbook_v2(WORKBOOK)

        # Start workflow.
        wf_ex = self.engine.start_workflow('wb1.with_items', WORKFLOW_INPUT)

        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')
        with_items_context = task1.runtime_context['with_items']

        self.assertEqual(3, with_items_context['count'])

        # Since we know that we can receive results in random order,
        # check is not depend on order of items.
        result = data_flow.get_task_execution_result(task1)

        self.assertTrue(isinstance(result, list))

        self.assertIn('John', result)
        self.assertIn('Ivan', result)
        self.assertIn('Mistral', result)

        published = task1.published

        self.assertIn(published['result'], ['John', 'Ivan', 'Mistral'])

        self.assertEqual(1, len(tasks))
        self.assertEqual(states.SUCCESS, task1.state)

    def test_with_items_static_var(self):
        wb_service.create_workbook_v2(WORKBOOK_WITH_STATIC_VAR)

        wf_input = copy.copy(WORKFLOW_INPUT)
        wf_input.update({'greeting': 'Hello'})
        # Start workflow.
        wf_ex = self.engine.start_workflow('wb1.with_items', wf_input)

        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.assertTrue(isinstance(result, list))

        self.assertIn('Hello, John!', result)
        self.assertIn('Hello, Ivan!', result)
        self.assertIn('Hello, Mistral!', result)

        self.assertEqual(1, len(tasks))
        self.assertEqual(states.SUCCESS, task1.state)

    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_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: <% task(task1).result %>
        """
        wf_service.create_workflows(wf)

        # Start workflow.
        wf_ex = self.engine.start_workflow('wf1_with_items', {})

        self.await_workflow_success(wf_ex.id)

        # 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([], result)

    def test_with_items_action_context(self):
        wb_service.create_workbook_v2(WB_ACTION_CONTEXT)

        # Start workflow.
        wf_ex = self.engine.start_workflow('wb.wf', WF_INPUT_URLS)

        with db_api.transaction():
            wf_ex = db_api.get_workflow_execution(wf_ex.id)

            task_ex = wf_ex.task_executions[0]

            act_exs = task_ex.executions

        self.engine.on_action_complete(act_exs[0].id, wf_utils.Result("Ivan"))
        self.engine.on_action_complete(act_exs[1].id, wf_utils.Result("John"))
        self.engine.on_action_complete(
            act_exs[2].id,
            wf_utils.Result("Mistral")
        )

        self.await_workflow_success(wf_ex.id)

        with db_api.transaction():
            task_ex = db_api.get_task_execution(task_ex.id)

            result = data_flow.get_task_execution_result(task_ex)

        self.assertIsInstance(result, list)

        self.assertIn('John', result)
        self.assertIn('Ivan', result)
        self.assertIn('Mistral', result)

        self.assertEqual(states.SUCCESS, task_ex.state)

    def test_with_items_multi_array(self):
        wb_service.create_workbook_v2(WORKBOOK_MULTI_ARRAY)

        wf_input = {'arrayI': ['a', 'b', 'c'], 'arrayJ': [1, 2, 3]}

        # Start workflow.
        wf_ex = self.engine.start_workflow('wb1.with_items', wf_input)

        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')

        # Since we know that we can receive results in random order,
        # check is not depend on order of items.
        result = data_flow.get_task_execution_result(task1)

        self.assertTrue(isinstance(result, list))

        self.assertIn('a 1', result)
        self.assertIn('b 2', result)
        self.assertIn('c 3', result)

        self.assertEqual(1, len(tasks))
        self.assertEqual(states.SUCCESS, task1.state)

    def test_with_items_results_one_item_as_list(self):
        wb_service.create_workbook_v2(WORKBOOK)

        # Start workflow.
        wf_ex = self.engine.start_workflow('wb1.with_items',
                                           WORKFLOW_INPUT_ONE_ITEM)

        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.assertTrue(isinstance(result, list))

        self.assertIn('Guy', result)

        published = task1.published

        self.assertIn(published['result'], ['Guy'])

        self.assertEqual(1, len(tasks))
        self.assertEqual(states.SUCCESS, task1.state)

    def test_with_items_concurrency_gt_list_length(self):
        workflow_definition = """---
        version: "2.0"

        concurrency_test:
          type: direct

          input:
           - names: ["John", "Ivan"]

          tasks:
            task1:
              with-items: name in <% $.names %>
              action: std.echo output=<% $.name %>
              concurrency: 3
        """

        wf_service.create_workflows(workflow_definition)

        # Start workflow.
        wf_ex = self.engine.start_workflow('concurrency_test', {})

        self._await(
            lambda: self.is_execution_success(wf_ex.id),
        )

        wf_ex = db_api.get_execution(wf_ex.id)
        task_ex = self._assert_single_item(wf_ex.task_executions, name='task1')
        result = data_flow.get_task_execution_result(task_ex)

        self.assertEqual(states.SUCCESS, task_ex.state)
        self.assertIsInstance(result, list)
        self.assertIn('John', result)
        self.assertIn('Ivan', result)

    def test_get_task_execution_result(self):
        task_ex = models.TaskExecution(
            name='task1',
            spec={
                "version": '2.0',
                'name': 'task1',
                'with-items': 'var in [1]',
                'type': 'direct'
            },
            runtime_context={
                'with_items_context': {'count': 1}
            }
        )

        action_exs = []

        action_exs.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=True,
            runtime_context={'with_items_index': 0}
        ))

        with mock.patch.object(db_api, 'get_action_executions',
                               return_value=action_exs):
            self.assertEqual([1], data_flow.get_task_execution_result(task_ex))

        action_exs.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=True,
            runtime_context={'with_items_index': 0}
        ))

        action_exs.append(models.ActionExecution(
            name='my_action',
            output={'result': 1},
            accepted=False,
            runtime_context={'with_items_index': 0}
        ))

        with mock.patch.object(db_api, 'get_action_executions',
                               return_value=action_exs):
            self.assertEqual(
                [1, 1],
                data_flow.get_task_execution_result(task_ex)
            )

    def test_with_items_subflow_concurrency_gt_list_length(self):
        wb_text = """---
        version: "2.0"
        name: wb1

        workflows:
          main:
            type: direct

            input:
             - names

            tasks:
              task1:
                with-items: name in <% $.names %>
                workflow: subflow1 name=<% $.name %>
                concurrency: 3

          subflow1:
            type: direct

            input:
                - name
            output:
              result: <% task(task1).result %>

            tasks:
              task1:
                action: std.echo output=<% $.name %>
        """

        wb_service.create_workbook_v2(wb_text)

        # Start workflow.
        names = ["Peter", "Susan", "Edmund", "Lucy", "Aslan", "Caspian"]
        wf_ex = self.engine.start_workflow('wb1.main', {'names': names})

        self.await_workflow_success(wf_ex.id)

        with db_api.transaction():
            wf_ex = db_api.get_workflow_execution(wf_ex.id)

            task_execs = wf_ex.task_executions

        task_ex = self._assert_single_item(
            task_execs,
            name='task1',
            state=states.SUCCESS
        )

        with db_api.transaction():
            task_ex = db_api.get_task_execution(task_ex.id)

            task_result = data_flow.get_task_execution_result(task_ex)

        result = [item['result'] for item in task_result]

        self.assertListEqual(sorted(result), sorted(names))

    def get(self, id):
        """Return the specified task."""
        LOG.info("Fetch task [id=%s]" % id)

        task_ex = db_api.get_task_execution(id)
        task = Task.from_dict(task_ex.to_dict())

        task.result = json.dumps(data_flow.get_task_execution_result(task_ex))

        return task