def test_rerun_with_items_concurrency(self):
        wb_service.create_workbook_v2(WITH_ITEMS_WORKBOOK_CONCURRENCY)

        # Run workflow and fail task.
        wf_ex = self.engine.start_workflow('wb3.wf1', {})
        self._await(lambda: self.is_execution_error(wf_ex.id))
        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertEqual(states.ERROR, wf_ex.state)
        self.assertIsNotNone(wf_ex.state_info)
        self.assertEqual(1, len(wf_ex.task_executions))

        task_1_ex = self._assert_single_item(wf_ex.task_executions, name='t1')

        self.assertEqual(states.ERROR, task_1_ex.state)

        task_1_action_exs = db_api.get_action_executions(
            task_execution_id=task_1_ex.id
        )

        self.assertEqual(4, len(task_1_action_exs))

        # Resume workflow and re-run failed task.
        self.engine.rerun_workflow(wf_ex.id, task_1_ex.id, reset=False)
        wf_ex = db_api.get_workflow_execution(wf_ex.id)

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

        self._await(lambda: self.is_execution_success(wf_ex.id), delay=10)
        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertEqual(states.SUCCESS, wf_ex.state)
        self.assertIsNone(wf_ex.state_info)
        self.assertEqual(2, len(wf_ex.task_executions))

        task_1_ex = self._assert_single_item(wf_ex.task_executions, name='t1')
        task_2_ex = self._assert_single_item(wf_ex.task_executions, name='t2')

        # Check action executions of task 1.
        self.assertEqual(states.SUCCESS, task_1_ex.state)
        self.assertIsNone(task_1_ex.state_info)

        task_1_action_exs = db_api.get_action_executions(
            task_execution_id=task_1_ex.id
        )

        # The action executions that succeeded should not re-run.
        self.assertEqual(6, len(task_1_action_exs))
        self.assertListEqual(['Task 1.0', 'Task 1.1', 'Task 1.2', 'Task 1.3'],
                             task_1_ex.published.get('v1'))

        # Check action executions of task 2.
        self.assertEqual(states.SUCCESS, task_2_ex.state)

        task_2_action_exs = db_api.get_action_executions(
            task_execution_id=task_2_ex.id
        )

        self.assertEqual(1, len(task_2_action_exs))

    def test_cancel_with_items_concurrency(self):
        wb_def = """
            version: '2.0'

            name: wb1

            workflows:
              wf1:
                tasks:
                  t1:
                    with-items: i in <% list(range(0, 4)) %>
                    action: std.async_noop
                    concurrency: 2
                    on-success:
                      - t2
                  t2:
                    action: std.echo output="Task 2"
        """

        wb_service.create_workbook_v2(wb_def)

        wf1_ex = self.engine.start_workflow('wb1.wf1')

        self.await_workflow_state(wf1_ex.id, states.RUNNING)

        with db_api.transaction():
            wf1_execs = db_api.get_workflow_executions()

            wf1_ex = self._assert_single_item(wf1_execs, name='wb1.wf1')
            wf1_t1_ex = self._assert_single_item(
                wf1_ex.task_executions,
                name='t1'
            )

        wf1_t1_action_exs = db_api.get_action_executions(
            task_execution_id=wf1_t1_ex.id
        )

        self.assertEqual(2, len(wf1_t1_action_exs))
        self.assertEqual(states.RUNNING, wf1_t1_action_exs[0].state)
        self.assertEqual(states.RUNNING, wf1_t1_action_exs[1].state)

        # Cancel action execution for task.
        for wf1_t1_action_ex in wf1_t1_action_exs:
            self.engine.on_action_complete(
                wf1_t1_action_ex.id,
                ml_actions.Result(cancel=True)
            )

        self.await_task_cancelled(wf1_t1_ex.id)
        self.await_workflow_cancelled(wf1_ex.id)

        wf1_t1_action_exs = db_api.get_action_executions(
            task_execution_id=wf1_t1_ex.id
        )

        self.assertEqual(2, len(wf1_t1_action_exs))
        self.assertEqual(states.CANCELLED, wf1_t1_action_exs[0].state)
        self.assertEqual(states.CANCELLED, wf1_t1_action_exs[1].state)

    def print_executions(exc_info):
        print("\nEngine test case exception occurred: %s" % exc_info[1])
        print("Exception type: %s" % exc_info[0])

        print("\nPrinting workflow executions...")

        with db_api.transaction():
            wf_execs = db_api.get_workflow_executions()

            for w in wf_execs:
                print(
                    "\n%s [state=%s, state_info=%s, output=%s]" %
                    (w.name, w.state, w.state_info, w.output)
                )

                for t in w.task_executions:
                    print(
                        "\t%s [id=%s, state=%s, state_info=%s, processed=%s,"
                        " published=%s]" %
                        (t.name,
                         t.id,
                         t.state,
                         t.state_info,
                         t.processed,
                         t.published)
                    )

                    a_execs = db_api.get_action_executions(
                        task_execution_id=t.id
                    )

                    for a in a_execs:
                        print(
                            "\t\t%s [id=%s, state=%s, state_info=%s,"
                            " accepted=%s, output=%s]" %
                            (a.name,
                             a.id,
                             a.state,
                             a.state_info,
                             a.accepted,
                             a.output)
                        )

        print("\nPrinting standalone action executions...")

        a_execs = db_api.get_action_executions(task_execution_id=None)

        for a in a_execs:
            print(
                "\t\t%s [id=%s, state=%s, state_info=%s, accepted=%s,"
                " output=%s]" %
                (a.name,
                 a.id,
                 a.state,
                 a.state_info,
                 a.accepted,
                 a.output)
            )

    def test_rerun_on_join_task(self):
        wb_service.create_workbook_v2(JOIN_WORKBOOK)

        # Run workflow and fail task.
        wf_ex = self.engine.start_workflow('wb1.wf1', {})
        wf_ex = db_api.get_workflow_execution(wf_ex.id)
        self._await(lambda: self.is_execution_error(wf_ex.id))
        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertEqual(states.ERROR, wf_ex.state)
        self.assertEqual(3, len(wf_ex.task_executions))

        task_1_ex = self._assert_single_item(wf_ex.task_executions, name='t1')
        task_2_ex = self._assert_single_item(wf_ex.task_executions, name='t2')
        task_3_ex = self._assert_single_item(wf_ex.task_executions, name='t3')

        self.assertEqual(states.SUCCESS, task_1_ex.state)
        self.assertEqual(states.SUCCESS, task_2_ex.state)
        self.assertEqual(states.ERROR, task_3_ex.state)

        # Resume workflow and re-run failed task.
        self.engine.rerun_workflow(wf_ex.id, task_3_ex.id)
        wf_ex = db_api.get_workflow_execution(wf_ex.id)

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

        # Wait for the workflow to succeed.
        self._await(lambda: self.is_execution_success(wf_ex.id))
        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertEqual(states.SUCCESS, wf_ex.state)
        self.assertEqual(3, len(wf_ex.task_executions))

        task_1_ex = self._assert_single_item(wf_ex.task_executions, name='t1')
        task_2_ex = self._assert_single_item(wf_ex.task_executions, name='t2')
        task_3_ex = self._assert_single_item(wf_ex.task_executions, name='t3')

        # Check action executions of task 1.
        task_1_action_exs = db_api.get_action_executions(
            task_execution_id=task_1_ex.id)

        self.assertEqual(1, len(task_1_action_exs))
        self.assertEqual(states.SUCCESS, task_1_action_exs[0].state)

        # Check action executions of task 2.
        task_2_action_exs = db_api.get_action_executions(
            task_execution_id=task_2_ex.id)

        self.assertEqual(1, len(task_2_action_exs))
        self.assertEqual(states.SUCCESS, task_2_action_exs[0].state)

        # Check action executions of task 3.
        task_3_action_exs = db_api.get_action_executions(
            task_execution_id=wf_ex.task_executions[2].id)

        self.assertEqual(2, len(task_3_action_exs))
        self.assertEqual(states.ERROR, task_3_action_exs[0].state)
        self.assertEqual(states.SUCCESS, task_3_action_exs[1].state)

    def test_next_task_with_input_yaql_error(self):
        wf_text = """
        version: '2.0'

        wf:
          type: direct

          tasks:
            task1:
              action: std.echo output="Echo"
              on-complete:
                - task2

            task2:
              action: std.echo output=<% wrong(yaql) %>
        """

        # Invoke workflow and assert workflow is in ERROR.
        wf_ex = self._run_workflow(wf_text)

        self.assertEqual(states.ERROR, wf_ex.state)
        self.assertIn('Can not evaluate YAQL expression', wf_ex.state_info)

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

            task_execs = wf_ex.task_executions

        self.assertEqual(2, len(task_execs))

        # 'task1' should be in SUCCESS.
        task_1_ex = self._assert_single_item(
            task_execs,
            name='task1',
            state=states.SUCCESS
        )

        # 'task1' should have exactly one action execution (in SUCCESS).
        task_1_action_exs = db_api.get_action_executions(
            task_execution_id=task_1_ex.id
        )

        self.assertEqual(1, len(task_1_action_exs))
        self.assertEqual(states.SUCCESS, task_1_action_exs[0].state)

        # 'task2' should exist but in ERROR.
        task_2_ex = self._assert_single_item(
            task_execs,
            name='task2',
            state=states.ERROR
        )

        # 'task2' must not have action executions.
        self.assertEqual(
            0,
            len(db_api.get_action_executions(task_execution_id=task_2_ex.id))
        )

    def test_join_all_task_with_input_jinja_error(self):
        wf_def = """---
        version: '2.0'
        wf:
          tasks:
            task_1_1:
              action: std.sleep seconds=1
              on-success:
                - task_2
            task_1_2:
              on-success:
                - task_2
            task_2:
              action: std.echo
              join: all
              input:
                output: |
                  !! {{ _.nonexistent_variable }} !!"""

        wf_service.create_workflows(wf_def)
        wf_ex = self.engine.start_workflow('wf')

        self.await_workflow_error(wf_ex.id)

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

        self.assertEqual(3, len(tasks))

        task_1_1 = self._assert_single_item(
            tasks, name="task_1_1", state=states.SUCCESS
        )
        task_1_2 = self._assert_single_item(
            tasks, name="task_1_2", state=states.SUCCESS
        )

        task_2 = self._assert_single_item(
            tasks, name="task_2", state=states.ERROR
        )

        with db_api.transaction():
            task_1_1_action_exs = db_api.get_action_executions(
                task_execution_id=task_1_1.id)
            task_1_2_action_exs = db_api.get_action_executions(
                task_execution_id=task_1_2.id)

            task_2_action_exs = db_api.get_action_executions(
                task_execution_id=task_2.id)

        self.assertEqual(1, len(task_1_1_action_exs))
        self.assertEqual(states.SUCCESS, task_1_1_action_exs[0].state)

        self.assertEqual(1, len(task_1_2_action_exs))
        self.assertEqual(states.SUCCESS, task_1_2_action_exs[0].state)

        self.assertEqual(0, len(task_2_action_exs))

    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 test_invalid_workflow_input(self):
        # Check that in case of invalid input workflow objects aren't even
        # created.
        wf_text = """
        version: '2.0'

        wf:
          input:
            - param1
            - param2

          tasks:
            task1:
              action: std.noop
        """

        wf_service.create_workflows(wf_text)

        self.assertRaises(
            exc.InputException,
            self.engine.start_workflow,
            'wf',
            '',
            {'wrong_param': 'some_value'}
        )

        self.assertEqual(0, len(db_api.get_workflow_executions()))
        self.assertEqual(0, len(db_api.get_task_executions()))
        self.assertEqual(0, len(db_api.get_action_executions()))

def get_task_execution_result(task_ex):
    # Use of task_ex.executions requires a session to lazy load the action
    # executions. This get_task_execution_result method is also invoked
    # from get_all in the task execution API controller. If there is a lot of
    # read against the API, it will lead to a lot of unnecessary DB locks
    # which result in possible deadlocks and WF execution failures. Therefore,
    # use db_api.get_action_executions here to avoid session-less use cases.
    action_execs = db_api.get_action_executions(task_execution_id=task_ex.id)
    action_execs.sort(
        key=lambda x: x.runtime_context.get('index')
    )

    results = [
        _extract_execution_result(ex)
        for ex in action_execs
        if hasattr(ex, 'output') and ex.accepted
    ]

    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

    def test_publish_failure(self):
        wb_service.create_workbook_v2(SIMPLE_WORKBOOK)

        # Run workflow and fail task.
        wf_ex = self.engine.start_workflow('wb1.wf1')

        self.await_workflow_error(wf_ex.id)

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

            task_execs = wf_ex.task_executions

        self.assertEqual(states.ERROR, wf_ex.state)
        self.assertEqual(1, len(task_execs))

        task_1_ex = self._assert_single_item(task_execs, name='t1')

        # Task 1 should have failed.
        self.assertEqual(states.ERROR, task_1_ex.state)
        self.assertIn('Can not evaluate YAQL expression', task_1_ex.state_info)

        # Action execution of task 1 should have succeeded.
        task_1_action_exs = db_api.get_action_executions(
            task_execution_id=task_1_ex.id
        )

        self.assertEqual(1, len(task_1_action_exs))
        self.assertEqual(states.SUCCESS, task_1_action_exs[0].state)

    def test_state_info_with_items(self):
        workflow = """---
        version: '2.0'
        wf:
          type: direct
          tasks:
            t1:
              with-items: i in <% list(range(0, 3)) %>
              action: std.echo output="Task 1.<% $.i %>"
        """

        wf_service.create_workflows(workflow)

        wf_ex = self.engine.start_workflow('wf', {})

        self.await_execution_error(wf_ex.id)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

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

        task_1_ex = self._assert_single_item(wf_ex.task_executions, name='t1')

        self.assertEqual(states.ERROR, task_1_ex.state)

        task_1_action_exs = db_api.get_action_executions(
            task_execution_id=task_1_ex.id)

        self.assertEqual(3, len(task_1_action_exs))
        self.assertIn(task_1_action_exs[0].id, wf_ex.state_info)
        self.assertNotIn(task_1_action_exs[1].id, wf_ex.state_info)
        self.assertIn(task_1_action_exs[2].id, wf_ex.state_info)

    def test_report_running_actions(self):
        wf_input = {'param1': 'Hey', 'param2': 'Hi'}

        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'wb.wf',
            '',
            wf_input=wf_input,
            description='my execution',
            task_name='task2'
        )

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

            task_execs = wf_ex.task_executions

        self.assertEqual(1, len(task_execs))

        task_ex = task_execs[0]

        action_execs = db_api.get_action_executions(
            task_execution_id=task_ex.id
        )

        task_action_ex = action_execs[0]

        self.engine.report_running_actions([])
        self.engine.report_running_actions([None, None])
        self.engine.report_running_actions([None, task_action_ex.id])

        task_action_ex = db_api.get_action_execution(task_action_ex.id)

        self.assertIsNotNone(task_action_ex.last_heartbeat)

def is_completed(task_ex):
    action_exs = db_api.get_action_executions(
        task_execution_id=task_ex.id,
        accepted=True
    )
    count = get_count(task_ex) or 1

    return count == len(action_exs)

    def test_second_task_with_input_jinja_error(self):
        wf_def = """---
        version: '2.0'
        wf:
          tasks:
            first:
              on-success:
                - second
            second:
              action: std.echo
              input:
                output: |
                  !! {{ _.nonexistent_variable }} !!"""

        wf_service.create_workflows(wf_def)
        wf_ex = self.engine.start_workflow('wf')

        self.await_workflow_error(wf_ex.id)

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

        self.assertEqual(2, len(tasks))

        task_first = self._assert_single_item(
            tasks, name="first", state=states.SUCCESS
        )
        task_second = self._assert_single_item(
            tasks, name="second", state=states.ERROR
        )

        with db_api.transaction():
            first_tasks_action_exs = db_api.get_action_executions(
                task_execution_id=task_first.id)
            second_tasks_action_exs = db_api.get_action_executions(
                task_execution_id=task_second.id)

        self.assertEqual(1, len(first_tasks_action_exs))
        self.assertEqual(states.SUCCESS, first_tasks_action_exs[0].state)

        self.assertEqual(0, len(second_tasks_action_exs))

def _get_action_executions(task_execution_id=None):
    kwargs = {'type': 'action_execution'}

    if task_execution_id:
        kwargs['task_execution_id'] = task_execution_id

    action_execs = [
        _get_action_execution_resource(a_ex)
        for a_ex in db_api.get_action_executions(**kwargs)
    ]

    return ActionExecutions(action_executions=action_execs)

    def test_cascade_delete(self):
        wf_text = """
        version: 2.0

        wf:
          tasks:
            task1:
              workflow: sub_wf1

            task2:
              workflow: sub_wf2

        sub_wf1:
          tasks:
            task1:
              action: std.noop

        sub_wf2:
          tasks:
            task1:
              action: std.noop
        """

        wf_service.create_workflows(wf_text)

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

        self.await_workflow_success(wf_ex.id)

        self.assertEqual(3, len(db_api.get_workflow_executions()))
        self.assertEqual(4, len(db_api.get_task_executions()))
        self.assertEqual(2, len(db_api.get_action_executions()))

        # Now delete the root workflow execution and make sure that
        # all dependent objects are deleted as well.
        db_api.delete_workflow_execution(wf_ex.id)

        self.assertEqual(0, len(db_api.get_workflow_executions()))
        self.assertEqual(0, len(db_api.get_task_executions()))
        self.assertEqual(0, len(db_api.get_action_executions()))

    def test_state_info_with_items(self):
        workflow = """---
        version: '2.0'
        wf:
          type: direct
          tasks:
            t1:
              with-items: i in <% list(range(0, 3)) %>
              action: std.echo output="Task 1.<% $.i %>"
        """

        wf_service.create_workflows(workflow)

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

        self.await_workflow_error(wf_ex.id)

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

            task_execs = wf_ex.task_executions

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

        task_1_ex = self._assert_single_item(task_execs, name='t1')

        self.assertEqual(states.ERROR, task_1_ex.state)

        task_1_action_exs = db_api.get_action_executions(
            task_execution_id=task_1_ex.id
        )

        self.assertEqual(3, len(task_1_action_exs))

        error_actions = [
            action_ex for action_ex in task_1_action_exs if
            action_ex.state == states.ERROR
        ]
        self.assertEqual(2, len(error_actions))

        success_actions = [
            action_ex for action_ex in task_1_action_exs if
            action_ex.state == states.SUCCESS
        ]
        self.assertEqual(1, len(success_actions))

        for action_ex in error_actions:
            self.assertIn(action_ex.id, wf_ex.state_info)

        for action_ex in success_actions:
            self.assertNotIn(action_ex.id, wf_ex.state_info)

    def test_on_action_update_non_async(self):
        workflow = """
        version: '2.0'
        wf_sync:
            type: direct
            tasks:
                task1:
                    action: std.noop
                    on-success:
                    - task2
                task2:
                    action: std.noop
        """

        # Start workflow.
        wf_service.create_workflows(workflow)
        wf_ex = self.engine.start_workflow('wf_sync')

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

        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

        self.assertEqual(1, len(task_execs))

        task1_ex = task_execs[0]

        self.assertEqual('task1', task1_ex.name)
        self.assertEqual(states.RUNNING, task1_ex.state)

        action_execs = db_api.get_action_executions(
            task_execution_id=task1_ex.id
        )

        self.assertEqual(1, len(action_execs))

        task1_action_ex = action_execs[0]

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

        self.assertRaises(
            exc.InvalidStateTransitionException,
            self.engine.on_action_update,
            task1_action_ex.id,
            states.PAUSED
        )

def _get_action_executions(task_execution_id=None):
    kwargs = {'type': 'action_execution'}

    if task_execution_id:
        kwargs['task_execution_id'] = task_execution_id

    action_executions = []

    for action_ex in db_api.get_action_executions(**kwargs):
        action_executions.append(
            _get_action_execution_resource(action_ex)
        )

    return ActionExecutions(action_executions=action_executions)

    def test_run_with_items(self):
        wb_def = """
        version: '2.0'

        name: wb1

        workflows:
          wf1:
            type: direct

            tasks:
              t1:
                with-items: i in <% list(range(0, 3)) %>
                action: std.echo output="Task 1.<% $.i %>"
                publish:
                  v1: <% task(t1).result %>
                on-success:
                  - t2
              t2:
                action: std.echo output="Task 2"
        """

        wb_svc.create_workbook_v2(wb_def)

        wf_ex = self.engine.start_workflow('wb1.wf1')

        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

        self.assertEqual(states.SUCCESS, wf_ex.state)
        self.assertEqual(2, len(wf_ex.task_executions))

        task_1_ex = self._assert_single_item(task_execs, name='t1')
        task_2_ex = self._assert_single_item(task_execs, name='t2')

        self.assertEqual(states.SUCCESS, task_1_ex.state)
        self.assertEqual(states.SUCCESS, task_2_ex.state)

        with db_api.transaction():
            task_1_action_exs = db_api.get_action_executions(
                task_execution_id=task_1_ex.id
            )

        self.assertEqual(3, len(task_1_action_exs))

        # Make sure the remote executor is not called.
        self.assertFalse(r_exe.RemoteExecutor.run_action.called)