def test_first_task_error(self):
        # Check that in case of an error in first task workflow objects are
        # still persisted properly.
        wf_text = """
        version: '2.0'

        wf:
          tasks:
            task1:
              action: std.fail
              on-success: task2

            task2:
              action: std.noop
        """

        wf_service.create_workflows(wf_text)

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

        self.assertEqual(states.RUNNING, wf_ex.state)
        self.assertIsNotNone(db_api.get_workflow_execution(wf_ex.id))

        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(1, len(task_execs))

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

    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_retry_policy_one_line(self):
        retry_wb = """---
        version: '2.0'

        name: wb

        workflows:
          wf1:
            type: direct

            tasks:
              task1:
                action: std.fail
                retry: count=3 delay=1
        """
        wb_service.create_workbook_v2(retry_wb)

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

        # Note: We need to reread execution to access related tasks.
        wf_ex = db_api.get_workflow_execution(wf_ex.id)
        task_ex = wf_ex.task_executions[0]

        self._await(lambda: self.is_task_error(task_ex.id))

        self._await(lambda: self.is_execution_error(wf_ex.id))

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

        self.assertEqual(
            2,
            task_ex.runtime_context['retry_task_policy']['retry_no']
        )

    def test_retry_policy_never_happen(self):
        retry_wb = """---
        version: '2.0'

        name: wb

        workflows:
          wf1:
            tasks:
              task1:
                action: std.echo output="hello"
                retry:
                  count: 3
                  delay: 1
        """
        wb_service.create_workbook_v2(retry_wb)

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

        # Note: We need to reread execution to access related tasks.
        wf_ex = db_api.get_workflow_execution(wf_ex.id)
        task_ex = wf_ex.task_executions[0]

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

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

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

        self.assertEqual(
            {},
            task_ex.runtime_context["retry_task_policy"]
        )

    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_pause_before_policy(self):
        wb_service.create_workbook_v2(PAUSE_BEFORE_WB)

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

        wf_ex = db_api.get_workflow_execution(wf_ex.id)
        task_ex = self._assert_single_item(
            wf_ex.task_executions,
            name='task1'
        )

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

        self._await(lambda: self.is_execution_paused(wf_ex.id))
        self._sleep(1)
        self.engine.resume_workflow(wf_ex.id)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)
        self._assert_single_item(wf_ex.task_executions, name='task1')

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

        wf_ex = db_api.get_workflow_execution(wf_ex.id)
        task_ex = self._assert_single_item(
            wf_ex.task_executions,
            name='task1'
        )
        next_task_ex = self._assert_single_item(
            wf_ex.task_executions,
            name='task2'
        )

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

    def test_resume_direct(self):
        wb_service.create_workbook_v2(RESUME_WORKBOOK)

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

        self._await(lambda: self.is_execution_paused(wf_ex.id))

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

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

        self.engine.resume_workflow(wf_ex.id)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertEqual(2, len(wf_ex.task_executions))

        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(2, len(wf_ex.task_executions))

    def test_resume_two_branches(self):
        wb_service.create_workbook_v2(WORKBOOK_TWO_BRANCHES)

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

        self.await_workflow_paused(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.PAUSED, wf_ex.state)
        self.assertEqual(3, len(task_execs))

        wf_ex = self.engine.resume_workflow(wf_ex.id)

        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)

        # We can see 3 tasks in execution.
        self.assertEqual(3, len(task_execs))

    def test_resume_two_start_tasks(self):
        wb_service.create_workbook_v2(WORKBOOK_TWO_START_TASKS)

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

        self._await(lambda: self.is_execution_paused(wf_ex.id))

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

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

        task_execs = wf_ex.task_executions

        # The exact number of tasks depends on which of two tasks
        # 'task1' and 'task2' completed earlier.
        self.assertTrue(len(task_execs) >= 2)

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

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

        self.engine.resume_workflow(wf_ex.id)

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

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

    def test_resume_direct(self):
        wb_service.create_workbook_v2(RESUME_WORKBOOK)

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

        self.await_workflow_paused(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.PAUSED, wf_ex.state)
        self.assertEqual(2, len(task_execs))

        self.engine.resume_workflow(wf_ex.id)

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

            self.assertEqual(2, len(wf_ex.task_executions))

        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(task_execs))

    def test_resume_reverse(self):
        wb_service.create_workbook_v2(RESUME_WORKBOOK_REVERSE)

        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'resume_reverse.wf',
            {},
            task_name='task2'
        )

        # Note: We need to reread execution to access related tasks.

        self.engine.pause_workflow(wf_ex.id)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertEqual(states.PAUSED, wf_ex.state)
        self.assertEqual(1, len(wf_ex.task_executions))

        self.engine.resume_workflow(wf_ex.id)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

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

        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(2, len(wf_ex.task_executions))

    def test_retry_policy_from_var(self):
        wb_service.create_workbook_v2(RETRY_WB_FROM_VAR)

        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'wb.wf1',
            wf_input={'count': 3, 'delay': 1}
        )

        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_ex = wf_ex.task_executions[0]

        self.assertEqual(states.RUNNING, task_ex.state)
        self.assertDictEqual({}, task_ex.runtime_context)

        self.await_task_delayed(task_ex.id, delay=0.5)
        self.await_task_error(task_ex.id)

        self.await_workflow_error(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.assertEqual(
            3,
            task_ex.runtime_context["retry_task_policy"]["retry_no"]
        )

    def test_long_action(self):
        wf_service.create_workflows(WF_LONG_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)
        self.assertEqual(states.RUNNING, wf_ex.task_executions[0].state)

        self.wait_for_action()

        # Here's the point when the action is blocked but already running.
        # Do the same check again, it should always pass.
        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertEqual(states.RUNNING, wf_ex.state)
        self.assertEqual(states.RUNNING, wf_ex.task_executions[0].state)

        self.unblock_action()

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

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        self.assertDictEqual({'result': 'test'}, wf_ex.output)

    def test_timeout_policy_success_after_timeout(self):
        wb_service.create_workbook_v2(TIMEOUT_WB2)

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

        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_ex = wf_ex.task_executions[0]

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

        self.await_task_error(task_ex.id)

        self.await_workflow_error(wf_ex.id)

        # Wait until timeout exceeds.
        self._sleep(1)

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

            task_execs = wf_ex.task_executions

        # Make sure that engine did not create extra tasks.
        self.assertEqual(1, len(task_execs))

    def test_retry_policy(self):
        wb_service.create_workbook_v2(RETRY_WB)

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

        # Note: We need to reread execution to access related tasks.
        wf_ex = db_api.get_workflow_execution(wf_ex.id)
        task_ex = wf_ex.task_executions[0]

        self.assertEqual(states.RUNNING, task_ex.state)
        self.assertDictEqual({}, task_ex.runtime_context)

        self._await(
            lambda: self.is_task_delayed(task_ex.id),
            delay=0.5
        )
        self._await(lambda: self.is_task_error(task_ex.id))

        self._await(lambda: self.is_execution_error(wf_ex.id))

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

        self.assertEqual(
            2,
            task_ex.runtime_context["retry_task_policy"]["retry_no"]
        )

    def test_subworkflow_root_execution_id(self):
        self.engine.start_workflow('wb6.wf1')

        self._await(lambda: len(db_api.get_workflow_executions()) == 3, 0.5, 5)

        wf_execs = db_api.get_workflow_executions()

        wf1_ex = self._assert_single_item(wf_execs, name='wb6.wf1')
        wf2_ex = self._assert_single_item(wf_execs, name='wb6.wf2')
        wf3_ex = self._assert_single_item(wf_execs, name='wb6.wf3')

        self.assertEqual(3, len(wf_execs))

        # Wait till workflow 'wf1' is completed (and all the sub-workflows
        # will be completed also).
        self.await_workflow_success(wf1_ex.id)

        with db_api.transaction():
            wf1_ex = db_api.get_workflow_execution(wf1_ex.id)
            wf2_ex = db_api.get_workflow_execution(wf2_ex.id)
            wf3_ex = db_api.get_workflow_execution(wf3_ex.id)

            self.assertIsNone(wf1_ex.root_execution_id, None)
            self.assertEqual(wf2_ex.root_execution_id, wf1_ex.id)
            self.assertEqual(wf2_ex.root_execution, wf1_ex)
            self.assertEqual(wf3_ex.root_execution_id, wf1_ex.id)
            self.assertEqual(wf3_ex.root_execution, wf1_ex)

    def test_cancel_parent_workflow(self):
        workbook = """
        version: '2.0'

        name: wb

        workflows:
            wf:
              type: direct
              tasks:
                taskx:
                  workflow: subwf

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

                task2:
                  action: std.echo output="foo"
                  wait-before: 2
        """

        wb_service.create_workbook_v2(workbook)

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

        self.engine.stop_workflow(
            wf_ex.id,
            states.CANCELLED,
            "Cancelled by user."
        )

        self.await_workflow_cancelled(wf_ex.id)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        task_ex = self._assert_single_item(wf_ex.task_executions, name='taskx')

        self.await_task_cancelled(task_ex.id)

        wf_ex = db_api.get_workflow_execution(wf_ex.id)

        task_ex = self._assert_single_item(wf_ex.task_executions, name='taskx')

        subwf_execs = db_api.get_workflow_executions(
            task_execution_id=task_ex.id
        )

        self.assertEqual(states.CANCELLED, wf_ex.state)
        self.assertEqual("Cancelled by user.", wf_ex.state_info)
        self.assertEqual(states.CANCELLED, task_ex.state)
        self.assertEqual("Cancelled by user.", task_ex.state_info)
        self.assertEqual(1, len(subwf_execs))
        self.assertEqual(states.CANCELLED, subwf_execs[0].state)
        self.assertEqual("Cancelled by user.", subwf_execs[0].state_info)

    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 _compute_delta(wf_ex):
            with db_api.transaction():
                # ensure that workflow execution exists
                db_api.get_workflow_execution(
                    id,
                    fields=(db_models.WorkflowExecution.id,)
                )

                delta = {}

                if wf_ex.state:
                    delta['state'] = wf_ex.state

                if wf_ex.description:
                    delta['description'] = wf_ex.description

                if wf_ex.params and wf_ex.params.get('env'):
                    delta['env'] = wf_ex.params.get('env')

                # Currently we can change only state, description, or env.
                if len(delta.values()) <= 0:
                    raise exc.InputException(
                        'The property state, description, or env '
                        'is not provided for update.'
                    )

                # Description cannot be updated together with state.
                if delta.get('description') and delta.get('state'):
                    raise exc.InputException(
                        'The property description must be updated '
                        'separately from state.'
                    )

                # If state change, environment cannot be updated
                # if not RUNNING.
                if (delta.get('env') and
                        delta.get('state') and
                        delta['state'] != states.RUNNING):
                    raise exc.InputException(
                        'The property env can only be updated when workflow '
                        'execution is not running or on resume from pause.'
                    )

                if delta.get('description'):
                    wf_ex = db_api.update_workflow_execution(
                        id,
                        {'description': delta['description']}
                    )

                if not delta.get('state') and delta.get('env'):
                    wf_ex = db_api.get_workflow_execution(id)
                    wf_ex = wf_service.update_workflow_execution_env(
                        wf_ex,
                        delta.get('env')
                    )

                return delta, wf_ex

    def test_task_execution_integrity(self):
        self.override_config('execution_integrity_check_delay', 1, 'engine')

        # The idea of the test is that we use the no-op asynchronous action
        # so that action and task execution state is not automatically set
        # to SUCCESS after we start the workflow. We'll update the action
        # execution state to SUCCESS directly through the DB and will wait
        # till task execution integrity is checked and fixed automatically
        # by a periodic job after about 2 seconds.
        wf_text = """
        version: '2.0'

        wf:
          tasks:
            task1:
              action: std.noop
              on-success: task2

            task2:
              action: std.async_noop
        """

        wf_service.create_workflows(wf_text)

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

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

            task1_ex = self._assert_single_item(
                wf_ex.task_executions,
                name='task1'
            )

        self.await_task_success(task1_ex.id)

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

            task2_ex = self._assert_single_item(
                wf_ex.task_executions,
                name='task2',
                state=states.RUNNING
            )
            action2_ex = self._assert_single_item(
                task2_ex.executions,
                state=states.RUNNING
            )

        db_api.update_action_execution(
            action2_ex.id,
            {'state': states.SUCCESS}
        )

        self.await_task_success(task2_ex.id)
        self.await_workflow_success(wf_ex.id)