def test_stop_workflow_fail(self):
        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'wb.wf', {'param1': 'Hey', 'param2': 'Hi'}, task_name="task2")
        # Re-read execution to access related tasks.
        wf_ex = db_api.get_execution(wf_ex.id)

        self.engine.stop_workflow(wf_ex.id, 'ERROR', "Stop this!")

        # Re-read from DB again
        wf_ex = db_api.get_execution(wf_ex.id)

        self.assertEqual('ERROR', wf_ex.state)
        self.assertEqual("Stop this!", wf_ex.state_info)

    def test_stop_workflow_succeed(self):
        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'wb.wf', {'param1': 'Hey', 'param2': 'Hi'}, task_name="task2")
        # Re-read execution to access related tasks.
        wf_ex = db_api.get_execution(wf_ex.id)

        self.engine.stop_workflow(wf_ex.id, 'SUCCESS', "Like this, done")

        # Re-read from DB again
        wf_ex = db_api.get_execution(wf_ex.id)

        self.assertEqual('SUCCESS', wf_ex.state)
        self.assertEqual("Like this, done", wf_ex.state_info)

    def test_delayed_task_and_correct_finish_workflow(self):
        wf_delayed_state = """---
        version: "2.0"
        wf:
          type: direct
          tasks:

            task1:
              action: std.noop
              wait-before: 1

            task2:
              action: std.noop
        """
        wf_service.create_workflows(wf_delayed_state)

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

        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_execution(wf_ex.id)

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

    def test_stop_succeeded(self):
        self.engine.stop_workflow(self.exec_id, states.ERROR, "Failure")

        self._await(lambda: self.is_execution_error(self.exec_id))
        wf_ex = db_api.get_execution(self.exec_id)
        self.assertEqual(states.ERROR, wf_ex.state)
        self.assertEqual("Failure", wf_ex.state_info)

    def test_stop_failed(self):
        self.engine.stop_workflow(self.exec_id, states.SUCCESS, "Force stop")

        self._await(lambda: self.is_execution_success(self.exec_id))
        wf_ex = db_api.get_execution(self.exec_id)
        self.assertEqual(states.SUCCESS, wf_ex.state)
        self.assertEqual("Force stop", wf_ex.state_info)

    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 stop_workflow(self, execution_id, state, message=None):
        with db_api.transaction():
            # Must be before loading the object itself (see method doc).
            self._lock_workflow_execution(execution_id)

            wf_ex = db_api.get_execution(execution_id)

            return self._stop_workflow(wf_ex, state, message)

    def test_stop_workflow_bad_status(self):
        wf_ex = self.engine.start_workflow(
            'wb.wf', {'param1': 'Hey', 'param2': 'Hi'}, task_name="task2")
        # Re-read execution to access related tasks.
        wf_ex = db_api.get_execution(wf_ex.id)

        self.assertNotEqual(
            'PAUSE',
            self.engine.stop_workflow(wf_ex.id, 'PAUSE')
        )

    def test_full_join_parallel_published_vars_complex(self):
        wfs_tasks_join_complex = """---
        version: "2.0"

        main:
          type: direct
          output:
            var_a: <% $.var_a %>
            var_b: <% $.var_b %>
            var_c: <% $.var_c %>
            var_d: <% $.var_d %>
          tasks:
            init:
              publish:
                var_a: 0
                var_b: 0
                var_c: 0
              on-success:
                - branch1_0
                - branch2_0

            branch1_0:
              publish:
                var_c: 1
              on-success:
                - branch1_1

            branch2_0:
              publish:
                var_a: 1
              on-success:
                - done

            branch1_1:
              publish:
                var_b: 1
              on-success:
                - done

            done:
              join: all
              publish:
                var_d: 1
        """
        wf_service.create_workflows(wfs_tasks_join_complex)

        # Start workflow.
        exec_db = self.engine.start_workflow("main", {})

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

        # Note: We need to reread execution to access related tasks.
        exec_db = db_api.get_execution(exec_db.id)

        self.assertDictEqual({"var_a": 1, "var_b": 1, "var_c": 1, "var_d": 1}, exec_db.output)

    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_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',
            state=states.SUCCESS
        )

        result = [
            item['result']
            for item in data_flow.get_task_execution_result(task_ex)
        ]

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

    def test_wait_before_policy_from_var(self):
        wb_service.create_workbook_v2(WAIT_BEFORE_FROM_VAR)

        # Start workflow.
        exec_db = self.engine.start_workflow('wb.wf1', {'wait_before': 1})

        # Note: We need to reread execution to access related tasks.
        exec_db = db_api.get_execution(exec_db.id)
        task_db = exec_db.task_executions[0]

        self.assertEqual(states.RUNNING_DELAYED, task_db.state)

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

    def test_expiration_policy_for_executions(self):
        # Delete execution uses a secured filtering and we need
        # to verify that admin able to do that for other projects.
        cfg.CONF.set_default('auth_enable', True, group='pecan')

        # Since we are removing other projects execution,
        # we want to load the executions with other project_id.
        _switch_context('non_admin_project', False)

        _load_executions()

        now = datetime.datetime.now()

        # This execution has a parent wf and testing that we are
        # querying only for parent wfs.
        exec_child = db_api.get_execution('654')

        self.assertEqual('789', exec_child.task_execution_id)

        # Call for all expired wfs execs.
        execs = db_api.get_expired_executions(now)

        # Should be only 3, the RUNNING execution shouldn't return,
        # so the child wf (that has parent task id).
        self.assertEqual(3, len(execs))

        # Switch context to Admin since expiration policy running as Admin.
        _switch_context(None, True)

        # TODO(m4dcoder): The expiration policy is changed here to expire
        # executions older than 30 minutes. It was originally 10 minutes.
        # The unit test below expects 1 execution to remain after the policy
        # is applied. However, the unit test fail frequently because the
        # process that deletes the expired executions seem to run late and
        # all executions are deleted. The unit tests seems to run better if
        # the config is changed to 30 minutes. Troubleshoot the expiration
        # policy to identify cause of the delay.
        _set_expiration_policy_config(1, 30)
        expiration_policy.run_execution_expiration_policy(self, ctx)

        # Only non_expired available (update_at < older_than).
        execs = db_api.get_expired_executions(now)

        self.assertEqual(1, len(execs))
        self.assertEqual('987', execs[0].id)

        _set_expiration_policy_config(1, 5)
        expiration_policy.run_execution_expiration_policy(self, ctx)
        execs = db_api.get_expired_executions(now)

        self.assertEqual(0, len(execs))

    def test_with_items_concurrency_2_fail(self):
        workflow_with_concurrency_2_fail = """---
        version: "2.0"

        concurrency_test_fail:
          type: direct

          tasks:
            task1:
              with-items: i in [1, 2, 3, 4]
              action: std.fail
              concurrency: 2
              on-error: task2

            task2:
              action: std.echo output="With-items failed"

        """
        wf_service.create_workflows(workflow_with_concurrency_2_fail)

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

        self._await(
            lambda: self.is_execution_success(wf_ex.id),
        )
        wf_ex = db_api.get_execution(wf_ex.id)

        task_exs = wf_ex.task_executions

        self.assertEqual(2, len(task_exs))

        task_2 = self._assert_single_item(task_exs, name='task2')

        self.assertEqual(
            "With-items failed",
            data_flow.get_task_execution_result(task_2)
        )

    def test_with_items_concurrency_3(self):
        workflow_with_concurrency_3 = """---
        version: "2.0"

        concurrency_test:
          type: direct

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

          tasks:
            task1:
              action: std.async_noop
              with-items: name in <% $.names %>
              concurrency: 3

        """
        wf_service.create_workflows(workflow_with_concurrency_3)

        # Start workflow.
        wf_ex = self.engine.start_workflow('concurrency_test', {})
        wf_ex = db_api.get_execution(wf_ex.id)
        task_ex = wf_ex.task_executions[0]

        self.assert_capacity(0, task_ex)
        self.assertEqual(3, self.get_running_action_exs_number(task_ex))

        # 1st iteration complete.
        self.engine.on_action_complete(
            self.get_incomplete_action_ex(task_ex).id,
            wf_utils.Result("John")
        )

        task_ex = db_api.get_task_execution(task_ex.id)
        self.assert_capacity(1, task_ex)

        # 2nd iteration complete.
        self.engine.on_action_complete(
            self.get_incomplete_action_ex(task_ex).id,
            wf_utils.Result("Ivan")
        )

        task_ex = db_api.get_task_execution(task_ex.id)
        self.assert_capacity(2, task_ex)

        # 3rd iteration complete.
        self.engine.on_action_complete(
            self.get_incomplete_action_ex(task_ex).id,
            wf_utils.Result("Mistral")
        )

        task_ex = db_api.get_task_execution(task_ex.id)
        self.assert_capacity(3, task_ex)

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

        task_ex = db_api.get_task_execution(task_ex.id)
        # 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(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 is_execution_in_state(self, ex_id, state):
     return db_api.get_execution(ex_id).state == state