def test_delete_workflow_completion_on_execution_delete(self):
        wf_text = """---
        version: '2.0'

        wf:
          tasks:
            async_task:
              action: std.async_noop
        """

        wf_service.create_workflows(wf_text)

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

        calls = db_api.get_delayed_calls()

        mtd_name = 'mistral.engine.workflow_handler._check_and_complete'

        self._assert_single_item(calls, target_method_name=mtd_name)

        db_api.delete_workflow_execution(wf_ex.id)

        self._await(
            lambda:
            len(db_api.get_delayed_calls(target_method_name=mtd_name)) == 0
        )

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

        wf:
          type: direct

          tasks:
            task1:
              action: std.echo output=1
              on-success: task2

            task2:
              action: std.echo output=1
              on-complete: task3

            task3:
              action: std.fail
              on-error: task4

            task4:
              action: std.echo output=4
        """

        wf_service.create_workflows(wf_text)

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

        self.await_workflow_success(wf_ex.id)

    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_with_items_action_defaults_from_env_not_applied(self):
        wf_service.create_workflows(WORKFLOW2_WITH_ITEMS)

        wf_input = {
            'links': [
                'https://api.library.org/books',
                'https://api.library.org/authors'
            ]
        }

        wf_ex = self.engine.start_workflow(
            'wf2_with_items',
            wf_input,
            env=ENV
        )

        self.await_workflow_success(wf_ex.id)

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

            self.assertEqual(states.SUCCESS, wf_ex.state)
            self._assert_single_item(wf_ex.task_executions, name='task1')

        calls = [mock.call('GET', url, params=None, data=None,
                           headers=None, cookies=None,
                           allow_redirects=None, proxies=None,
                           auth=EXPECTED_ENV_AUTH, verify=None,
                           timeout=60)
                 for url in wf_input['links']]

        requests.request.assert_has_calls(calls, any_order=True)

    def test_workflow_input_default_value_limit(self):
        new_wf = generate_workflow(['__WORKFLOW_INPUT__'])

        wf_service.create_workflows(new_wf)

        # Start workflow.
        self.engine.start_workflow('wf', {})

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

        wf:
          tasks:
            task1:
              action: std.fail
              wait-before: 2
        """

        wf_service.create_workflows(wf_text)

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

        self.await_workflow_error(wf_ex.id)

        task_ex = self._extract_task_ex(wf_ex.id)

        started_1st, finished_1st = self._get_started_finished(task_ex)

        # Make sure to rerun the workflow after a certain delay so that
        # times for the first run are different from times in the second run.
        eventlet.sleep(1)

        wf_ex = self.engine.rerun_workflow(task_ex.id)

        self.await_workflow_error(wf_ex.id)

        task_ex = self._extract_task_ex(wf_ex.id)

        started_2nd, finished_2nd = self._get_started_finished(task_ex)

        self.assertNotEqual(started_1st, started_2nd)
        self.assertNotEqual(finished_1st, finished_2nd)

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

        wf:
          tasks:
            task1:
              action: std.noop
              on-success:
                - succeed: <% $.invalid_yaql %>
        """

        wf_service.create_workflows(wf_text)

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

        state_info = task_ex.state_info

        self.assertIsNotNone(state_info)
        self.assertLess(state_info.find('error'), state_info.find('data'))

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

        wf:
          tasks:
            task1:
              action: std.noop
              publish:
                my_var: <% invalid_yaql_function() %>
        """

        wf_service.create_workflows(wf_text)

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

        state_info = task_ex.state_info

        self.assertIsNotNone(state_info)
        self.assertGreater(state_info.find('error='), 0)
        self.assertLess(state_info.find('error='), state_info.find('data='))

    def test_action_error(self):
        # Check that state of all workflow objects (workflow executions,
        # task executions, action executions) is properly persisted in case
        # of action error.
        wf_text = """
        version: '2.0'

        wf:
          tasks:
            task1:
              action: std.fail
        """

        wf_service.create_workflows(wf_text)

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

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

    def test_task_error_with_on_handlers(self):
        # Check that state of all workflow objects (workflow executions,
        # task executions, action executions) is properly persisted in case
        # of an error at task level and this task has on-XXX handlers.
        wf_text = """
        version: '2.0'

        wf:
          tasks:
            task1:
              action: std.noop
              publish:
                my_var: <% invalid_yaql_function() %>
              on-success:
                - task2
              on-error:
                - task3

            task2:
              description: This task must never run.
              action: std.noop

            task3:
              action: std.noop
        """

        wf_service.create_workflows(wf_text)

        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)

            # Now we need to make sure that task is in ERROR state but action
            # is in SUCCESS because error occurred in 'publish' clause which
            # must not affect action state.
            task_execs = wf_ex.task_executions

            # NOTE: task3 must not run because on-error handler triggers
            # only on error outcome of an action (or workflow) associated
            # with a task.
            self.assertEqual(1, len(task_execs))

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

            action_execs = task_ex.executions

        self.assertEqual(1, len(action_execs))

        self._assert_single_item(
            action_execs,
            name='std.noop',
            state=states.SUCCESS
        )

    def test_cancel_completed_workflow(self):
        workflow = """
        version: '2.0'

        wf:
          type: direct
          tasks:
            task1:
              action: std.echo output="Echo"
        """

        wf_service.create_workflows(workflow)

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

        self.await_workflow_success(wf_ex.id)

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

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

            task_execs = wf_ex.task_executions

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

        self.assertEqual(states.SUCCESS, wf_ex.state)
        self.assertIsNone(wf_ex.state_info)
        self.assertEqual(1, len(task_execs))
        self.assertEqual(states.SUCCESS, task_1_ex.state)

    def test_async_success_result(self):
        wf_service.create_workflows(WF.format(action_name="my_async_action"))

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

        # When the action is successful, the workflow will wait in the RUNNING
        # state for it to complete.
        self.await_workflow_running(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(1, len(tasks))

            task1 = self._assert_single_item(tasks, name='task1')
            self.assertEqual(states.RUNNING, task1.state)

    def test_task_published_limit(self):
        new_wf = generate_workflow(['__TASK_PUBLISHED__'])

        wf_service.create_workflows(new_wf)

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

        self.await_workflow_error(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

        self.assertIn(
            'Failed to handle action completion [error=Size of',
            wf_ex.state_info
        )
        self.assertIn('wf=wf, task=task1', wf_ex.state_info)

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

        self.assertIn(
            "Size of 'published' is 1KB which exceeds the limit of 0KB",
            task_ex.state_info
        )

    def test_db_error_in_jinja_expression(self):
        # This test just checks that the workflow completes successfully
        # even if a DB deadlock occurs during Jinja expression evaluation.
        # The engine in this case should should just retry the transactional
        # method.
        wf_text = """---
        version: '2.0'

        wf:
          tasks:
            task1:
              action: std.echo output="Hello"
              publish:
                my_var: "{{ 1 + 1 }}"
        """

        wf_service.create_workflows(wf_text)

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

        self.await_workflow_success(wf_ex.id)

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

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

            task_ex = wf_ex.task_executions[0]

            self.assertDictEqual({'my_var': 2}, task_ex.published)

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

        wf:
          tasks:
            task1:
              action: std.noop
              wait-before: 1
              wait-after: 2
        """

        wf_service.create_workflows(wf_text)

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

        self.await_workflow_success(wf_ex.id)

        task_ex = self._extract_task_ex(wf_ex.id)

        started, finished = self._get_started_finished(task_ex)

        duration = self._get_task_duration(started, finished)

        self._check_duration_more_than(duration, 1)

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

        wf:
          tasks:
            task1:
              action: std.echo output={{ _.invalid_var }}
              wait-before: 1
        """

        wf_service.create_workflows(wf_text)

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

        state_info = task_ex.state_info

        self.assertIsNotNone(state_info)
        self.assertGreater(state_info.find('error='), 0)
        self.assertLess(state_info.find('error='), state_info.find('wf='))

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

        wf:
          tasks:
            task1:
              action: std.fail
              retry:
                delay: 1
                count: 5
        """

        wf_service.create_workflows(wf_text)

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

        self.await_workflow_error(wf_ex.id)

        task_ex = self._extract_task_ex(wf_ex.id)

        created_at = task_ex.created_at
        started_at = self._get_started_finished(task_ex)[0]

        self.assertEqual(created_at, started_at)

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

        wf:
          tasks:
            task1:
              action: std.noop
              wait-after: 1
              on-success:
                - task2: <% invalid_yaql_function() %>

            task2:
              action: std.noop
        """

        wf_service.create_workflows(wf_text)

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

        state_info = task_ex.state_info

        self.assertIsNotNone(state_info)
        self.assertGreater(state_info.find('error='), 0)
        self.assertLess(state_info.find('error='), state_info.find('wf='))