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 _fail_task_if_incomplete(task_ex_id, timeout):
    from mistral.engine import task_handler

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

        if not states.is_completed(task_ex.state):
            msg = 'Task timed out [timeout(s)=%s].' % timeout

            task_handler.complete_task(
                db_api.get_task_execution(task_ex_id),
                states.ERROR,
                msg
            )

    def put(self, id, task):
        """Update the specified task execution.

        :param id: Task execution ID.
        :param task: Task execution object.
        """
        acl.enforce('tasks:update', context.ctx())

        LOG.info("Update task execution [id=%s, task=%s]" % (id, task))

        task_ex = db_api.get_task_execution(id)
        task_spec = spec_parser.get_task_spec(task_ex.spec)
        task_name = task.name or None
        reset = task.reset
        env = task.env or None

        if task_name and task_name != task_ex.name:
            raise exc.WorkflowException('Task name does not match.')

        wf_ex = db_api.get_workflow_execution(task_ex.workflow_execution_id)
        wf_name = task.workflow_name or None

        if wf_name and wf_name != wf_ex.name:
            raise exc.WorkflowException('Workflow name does not match.')

        if task.state != states.RUNNING:
            raise exc.WorkflowException(
                'Invalid task state. Only updating task to rerun is supported.'
            )

        if task_ex.state != states.ERROR:
            raise exc.WorkflowException(
                'The current task execution must be in ERROR for rerun.'
                ' Only updating task to rerun is supported.'
            )

        if not task_spec.get_with_items() and not reset:
            raise exc.WorkflowException(
                'Only with-items task has the option to not reset.'
            )

        rpc.get_engine_client().rerun_workflow(
            task_ex.id,
            reset=reset,
            env=env
        )

        task_ex = db_api.get_task_execution(id)

        return _get_task_resource_with_result(task_ex)

def analyse_task_execution(task_ex_id, stat, filters, cur_depth):
    with db_api.transaction():
        task_ex = db_api.get_task_execution(task_ex_id)

        if filters['errors_only'] and task_ex.state != states.ERROR:
            return None

        update_statistics_with_task(stat, task_ex)

        entry = create_task_execution_entry(task_ex)

        child_executions = task_ex.executions

    entry.action_executions = []
    entry.workflow_executions = []

    for c_ex in child_executions:
        if isinstance(c_ex, db_models.ActionExecution):
            entry.action_executions.append(
                create_action_execution_entry(c_ex)
            )
        else:
            entry.workflow_executions.append(
                analyse_workflow_execution(c_ex.id, stat, filters, cur_depth)
            )

    return entry

    def test_short_action(self):
        wf_service.create_workflows(WF_SHORT_ACTION)

        self.block_action()

        wf_ex = self.engine.start_workflow('wf', None)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertEqual(states.RUNNING, wf_ex.state)

        task_execs = wf_ex.task_executions

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

        self._await(lambda: self.is_task_success(task1_ex.id))

        self.unblock_action()

        self._await(lambda: self.is_task_success(task2_ex.id))
        self._await(lambda: self.is_execution_success(wf_ex.id))

        task1_ex = db_api.get_task_execution(task1_ex.id)
        task1_action_ex = db_api.get_action_executions(
            task_execution_id=task1_ex.id
        )[0]

        self.assertEqual(1, task1_action_ex.output['result'])

    def on_task_state_change(self, task_ex_id, state):
        with db_api.transaction():
            task_ex = db_api.get_task_execution(task_ex_id)
            # TODO(rakhmerov): The method is mostly needed for policy and
            # we are supposed to get the same action execution as when the
            # policy worked. But by the moment this method is called the
            # last execution object may have changed. It's a race condition.
            execution = task_ex.executions[-1]

            wf_ex_id = task_ex.workflow_execution_id

            # Must be before loading the object itself (see method doc).
            self._lock_workflow_execution(wf_ex_id)

            wf_ex = task_ex.workflow_execution

            wf_trace.info(
                task_ex,
                "Task '%s' [%s -> %s]"
                % (task_ex.name, task_ex.state, state)
            )

            task_ex.state = state

            self._on_task_state_change(task_ex, wf_ex, action_ex=execution)

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

        task_ex = db_api.get_task_execution(id)

        return _get_task_resource_with_result(task_ex)

def _complete_task(task_ex_id, state, state_info):
    from mistral.engine import task_handler

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

        task_handler.complete_task(task_ex, state, state_info)

def task_(context, task_name=None):
    # This section may not exist in a context if it's calculated not in
    # task scope.
    cur_task = context['__task_execution']

    # 1. If task_name is empty it's 'task()' use case, we need to get the
    # current task.
    # 2. if task_name is not empty but it's equal to the current task name
    # we need to take exactly the current instance of this task. Otherwise
    # there may be ambiguity if there are many tasks with this name.
    # 3. In other case we just find a task in DB by the given name.
    if cur_task and (not task_name or cur_task['name'] == task_name):
        task_ex = db_api.get_task_execution(cur_task['id'])
    else:
        task_execs = db_api.get_task_executions(
            workflow_execution_id=context['__execution']['id'],
            name=task_name
        )

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

    if not task_ex:
        LOG.warning(
            "Task '%s' not found by the task() expression function",
            task_name
        )
        return None

    # We don't use to_dict() db model method because not all fields
    # make sense for user.
    return _convert_to_user_model(task_ex)

    def _get_induced_join_state(self, in_task_spec, in_task_ex,
                                join_task_spec, t_execs_cache):
        join_task_name = join_task_spec.get_name()

        if not in_task_ex:
            possible, depth = self._possible_route(
                in_task_spec,
                t_execs_cache
            )

            if possible:
                return states.WAITING, depth, None
            else:
                return states.ERROR, depth, 'impossible route'

        if not states.is_completed(in_task_ex.state):
            return states.WAITING, 1, None

        if self._is_conditional_transition(in_task_ex, in_task_spec) and \
                not hasattr(in_task_ex, "in_context"):
            in_task_ex = db_api.get_task_execution(in_task_ex.id)

        # [(task name, params, event name), ...]
        next_tasks_tuples = self._find_next_tasks(in_task_ex)

        next_tasks_dict = {tup[0]: tup[2] for tup in next_tasks_tuples}

        if join_task_name not in next_tasks_dict:
            return states.ERROR, 1, "not triggered"

        return states.RUNNING, 1, next_tasks_dict[join_task_name]

    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 _recursive_rerun(self):
        """Rerun all parent workflow executions recursively.

        If there is a parent execution that it reruns as well.
        """

        from mistral.engine import workflow_handler

        self.set_state(states.RUNNING)

        # TODO(rakhmerov): We call a internal method of a module here.
        # The simplest way is to make it public, however, I believe
        # it's another "bad smell" that tells that some refactoring
        # of the architecture should be made.
        workflow_handler._schedule_check_and_fix_integrity(self.wf_ex)

        if self.wf_ex.task_execution_id:
            parent_task_ex = db_api.get_task_execution(
                self.wf_ex.task_execution_id
            )

            parent_wf = Workflow(wf_ex=parent_task_ex.workflow_execution)

            parent_wf.lock()

            parent_wf._recursive_rerun()

            from mistral.engine import task_handler
            task_handler.rerun_task(parent_task_ex, parent_wf.wf_spec)

    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 get(self, id):
        """Return the specified task."""
        acl.enforce('tasks:get', context.ctx())
        LOG.info("Fetch task [id=%s]" % id)

        task_ex = db_api.get_task_execution(id)

        return _get_task_resource_with_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 rerun_workflow(self, task_ex_id, reset=True, env=None):
        with db_api.transaction():
            task_ex = db_api.get_task_execution(task_ex_id)

            wf_ex = task_ex.workflow_execution

            wf_handler.rerun_workflow(wf_ex, task_ex, reset=reset, env=env)

            return wf_ex.get_clone()

    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

def fail_task_if_incomplete(task_ex_id, timeout):
    task_ex = db_api.get_task_execution(task_ex_id)

    if not states.is_completed(task_ex.state):
        msg = "Task timed out [id=%s, timeout(s)=%s]." % (task_ex_id, timeout)

        wf_trace.info(task_ex, msg)

        wf_trace.info(task_ex, "Task '%s' [%s -> ERROR]" % (task_ex.name, task_ex.state))

        rpc.get_engine_client().on_task_state_change(task_ex_id, states.ERROR)

    def test_retry_async_action(self):
        retry_wf = """---
          version: '2.0'
          repeated_retry:
            tasks:
              async_http:
                retry:
                  delay: 0
                  count: 100
                action: std.mistral_http url='https://google.com'
            """

        wf_service.create_workflows(retry_wf)
        wf_ex = self.engine.start_workflow('repeated_retry')

        self.await_workflow_running(wf_ex.id)

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

            task_ex = wf_ex.task_executions[0]
            self.await_task_running(task_ex.id)

            first_action_ex = task_ex.executions[0]
            self.await_action_state(first_action_ex.id, states.RUNNING)

        complete_action_params = (
            first_action_ex.id,
            ml_actions.Result(error="mock")
        )
        rpc.get_engine_client().on_action_complete(*complete_action_params)

        for _ in range(2):
            self.assertRaises(
                exc.MistralException,
                rpc.get_engine_client().on_action_complete,
                *complete_action_params
            )

        self.await_task_running(task_ex.id)
        with db_api.transaction():
            task_ex = db_api.get_task_execution(task_ex.id)
            action_exs = task_ex.executions

            self.assertEqual(2, len(action_exs))

            for action_ex in action_exs:
                if action_ex.id == first_action_ex.id:
                    expected_state = states.ERROR
                else:
                    expected_state = states.RUNNING

                self.assertEqual(expected_state, action_ex.state)

    def test_pause_before_with_delay_policy(self):
        wb_service.create_workbook_v2(PAUSE_BEFORE_DELAY_WB)

        # Start workflow.
        wf_ex = self.engine.start_workflow('wb.wf1')

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

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

        # Verify wf paused by pause-before
        self.await_workflow_paused(wf_ex.id)

        # Allow wait-before to expire
        self._sleep(2)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        # Verify wf still paused (wait-before didn't reactivate)
        self.await_workflow_paused(wf_ex.id)

        task_ex = db_api.get_task_execution(task_ex.id)

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

        self.engine.resume_workflow(wf_ex.id)

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

            task_execs = wf_ex.task_executions

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

        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')
        next_task_ex = self._assert_single_item(task_execs, name='task2')

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