def query(self, execution_id, query_context, last_query_time=None):
        """
        Queries mistral for workflow results using v2 APIs.
        :param execution_id: st2 execution_id (context to be used for logging/audit)
        :type execution_id: ``str``
        :param query_context: context for the query to be made to mistral. This contains mistral
                              execution id.
        :type query_context: ``object``
        :param last_query_time: Timestamp of last query.
        :type last_query_time: ``float``
        :rtype: (``str``, ``object``)
        """
        # Retrieve liveaction_db to append new result to existing result.
        liveaction_db = action_utils.get_liveaction_by_id(execution_id)

        mistral_exec_id = query_context.get('mistral', {}).get('execution_id', None)
        if not mistral_exec_id:
            raise Exception('[%s] Missing mistral workflow execution ID in query context. %s'
                            % (execution_id, query_context))

        LOG.info('[%s] Querying mistral execution %s...', execution_id, mistral_exec_id)

        try:
            wf_result = self._get_workflow_result(execution_id, mistral_exec_id)

            stream = getattr(liveaction_db, 'result', {})

            wf_tasks_result = self._get_workflow_tasks(
                execution_id,
                mistral_exec_id,
                recorded_tasks=stream.get('tasks', [])
            )

            result = self._format_query_result(
                liveaction_db.result,
                wf_result,
                wf_tasks_result
            )
        except exceptions.ReferenceNotFoundError as exc:
            LOG.exception('[%s] Unable to find reference.', execution_id)
            return (action_constants.LIVEACTION_STATUS_FAILED, str(exc))
        except Exception:
            LOG.exception('[%s] Unable to fetch mistral workflow result and tasks. %s',
                          execution_id, query_context)
            raise

        # Retrieve liveaction_db again in case state has changed
        # while the querier get results from mistral API above.
        liveaction_db = action_utils.get_liveaction_by_id(execution_id)

        status = self._determine_execution_status(
            liveaction_db,
            result['extra']['state'],
            result['tasks']
        )

        LOG.info('[%s] Determined execution status: %s', execution_id, status)
        LOG.debug('[%s] Combined execution result: %s', execution_id, result)

        return (status, result)

    def test_execute_cancelation(self):
        liveaction_db = self._create_liveaction_db()
        self._process_request(liveaction_db)

        scheduled_liveaction_db = action_db.get_liveaction_by_id(liveaction_db.id)
        scheduled_liveaction_db = self._wait_on_status(
            scheduled_liveaction_db,
            action_constants.LIVEACTION_STATUS_SCHEDULED
        )

        action_db.update_liveaction_status(
            status=action_constants.LIVEACTION_STATUS_CANCELED,
            liveaction_id=liveaction_db.id
        )

        canceled_liveaction_db = action_db.get_liveaction_by_id(liveaction_db.id)
        self.dispatcher._queue_consumer._process_message(canceled_liveaction_db)
        dispatched_liveaction_db = action_db.get_liveaction_by_id(liveaction_db.id)

        self.assertEqual(
            dispatched_liveaction_db.status,
            action_constants.LIVEACTION_STATUS_CANCELED
        )

        self.assertDictEqual(
            dispatched_liveaction_db.result,
            {'message': 'Action execution canceled by user.'}
        )

    def test_execute_no_result(self):
        live_action_db = self._get_execution_db_model(
            status=action_constants.LIVEACTION_STATUS_REQUESTED)

        self.scheduler._queue_consumer._process_message(live_action_db)
        scheduled_live_action_db = action_db.get_liveaction_by_id(live_action_db.id)
        self.assertEqual(scheduled_live_action_db.status,
                         action_constants.LIVEACTION_STATUS_SCHEDULED)

        self.dispatcher._queue_consumer._process_message(scheduled_live_action_db)
        dispatched_live_action_db = action_db.get_liveaction_by_id(live_action_db.id)
        self.assertEqual(dispatched_live_action_db.status,
                         action_constants.LIVEACTION_STATUS_FAILED)

    def test_execute_no_result(self):
        liveaction_db = self._create_liveaction_db()
        self._process_request(liveaction_db)

        scheduled_liveaction_db = action_db.get_liveaction_by_id(liveaction_db.id)
        scheduled_liveaction_db = self._wait_on_status(
            scheduled_liveaction_db,
            action_constants.LIVEACTION_STATUS_SCHEDULED
        )

        self.dispatcher._queue_consumer._process_message(scheduled_liveaction_db)
        dispatched_liveaction_db = action_db.get_liveaction_by_id(liveaction_db.id)
        self.assertEqual(dispatched_liveaction_db.status, action_constants.LIVEACTION_STATUS_FAILED)

    def test_execute(self):
        live_action_db = self._get_execution_db_model(
            status=action_constants.LIVEACTION_STATUS_REQUESTED)

        self.scheduler._queue_consumer._process_message(live_action_db)
        scheduled_live_action_db = action_db.get_liveaction_by_id(live_action_db.id)
        self.assertDictEqual(scheduled_live_action_db.runner_info, {})
        self.assertEqual(scheduled_live_action_db.status,
                         action_constants.LIVEACTION_STATUS_SCHEDULED)

        self.dispatcher._queue_consumer._process_message(scheduled_live_action_db)
        dispatched_live_action_db = action_db.get_liveaction_by_id(live_action_db.id)
        self.assertGreater(len(dispatched_live_action_db.runner_info.keys()), 0)
        self.assertEqual(dispatched_live_action_db.status,
                         action_constants.LIVEACTION_STATUS_RUNNING)

    def _run_action(self, action_node, parent_execution_id, params, wait_for_completion=True):
        liveaction = LiveActionDB(action=action_node.ref)
        liveaction.parameters = action_param_utils.cast_params(action_ref=action_node.ref,
                                                               params=params)

        # Setup notify for task in chain.
        notify = self._get_notify(action_node)
        if notify:
            liveaction.notify = notify
            LOG.debug('%s: Task notify set to: %s', action_node.name, liveaction.notify)

        liveaction.context = {
            'parent': str(parent_execution_id),
            'chain': vars(action_node)
        }

        liveaction, _ = action_service.request(liveaction)

        while (wait_for_completion and
               liveaction.status != LIVEACTION_STATUS_SUCCEEDED and
               liveaction.status != LIVEACTION_STATUS_FAILED):
            eventlet.sleep(1)
            liveaction = action_db_util.get_liveaction_by_id(liveaction.id)

        return liveaction

    def _update_live_action_db(self, liveaction_id, status, result, context):
        """
        Update LiveActionDB object for the provided liveaction id.
        """
        liveaction_db = get_liveaction_by_id(liveaction_id)

        state_changed = (
            liveaction_db.status != status and
            liveaction_db.status not in action_constants.LIVEACTION_COMPLETED_STATES
        )

        if status in action_constants.LIVEACTION_COMPLETED_STATES:
            end_timestamp = date_utils.get_datetime_utc_now()
        else:
            end_timestamp = None

        liveaction_db = update_liveaction_status(
            status=status if state_changed else liveaction_db.status,
            result=result,
            context=context,
            end_timestamp=end_timestamp,
            liveaction_db=liveaction_db
        )

        return (liveaction_db, state_changed)

 def test_succeeded_execution_handling(self):
     testworker = worker.Worker(None)
     live_action_db = self._get_execution_db_model()
     testworker.execute_action(live_action_db)
     updated_live_action_db = get_liveaction_by_id(live_action_db.id)
     self.assertEqual(updated_live_action_db.status,
                      action_constants.LIVEACTION_STATUS_RUNNING)

    def run(self, action_parameters):
        liveaction_db = action_utils.get_liveaction_by_id(self.liveaction_id)
        exc = ex_db_access.ActionExecution.get(liveaction__id=str(liveaction_db.id))

        # Assemble and dispatch trigger
        trigger_ref = sys_db_models.ResourceReference.to_string_reference(
            pack=trigger_constants.INQUIRY_TRIGGER['pack'],
            name=trigger_constants.INQUIRY_TRIGGER['name']
        )

        trigger_payload = {
            "id": str(exc.id),
            "route": self.route
        }

        self.trigger_dispatcher.dispatch(trigger_ref, trigger_payload)

        result = {
            "schema": self.schema,
            "roles": self.roles_param,
            "users": self.users_param,
            "route": self.route,
            "ttl": self.ttl
        }

        return (action_constants.LIVEACTION_STATUS_PENDING, result, None)

def abandon_execution_if_incomplete(liveaction_id, publish=True):
    """
    Marks execution as abandoned if it is still incomplete. Abandoning an
    execution implies that its end state is unknown and cannot anylonger
    be determined. This method should only be called if the owning process
    is certain it can no longer determine status of an execution.
    """
    liveaction_db = action_utils.get_liveaction_by_id(liveaction_id)

    # No need to abandon and already complete action
    if liveaction_db.status in action_constants.LIVEACTION_COMPLETED_STATES:
        raise ValueError('LiveAction %s already in a completed state %s.' %
                         (liveaction_id, liveaction_db.status))

    # Update status to reflect execution being abandoned.
    liveaction_db = action_utils.update_liveaction_status(
        status=action_constants.LIVEACTION_STATUS_ABANDONED,
        liveaction_db=liveaction_db,
        result={})

    execution_db = update_execution(liveaction_db, publish=publish)

    LOG.info('Marked execution %s as %s.', execution_db.id,
             action_constants.LIVEACTION_STATUS_ABANDONED)

    # Invoke post run on the action to execute post run operations such as callback.
    runners_utils.invoke_post_run(liveaction_db)

    return execution_db

    def run(self, action_parameters):

        liveaction_db = action_utils.get_liveaction_by_id(self.liveaction_id)
        exc = ActionExecution.get(liveaction__id=str(liveaction_db.id))

        # Assemble and dispatch trigger
        trigger_ref = ResourceReference.to_string_reference(
            pack=INQUIRY_TRIGGER['pack'],
            name=INQUIRY_TRIGGER['name']
        )
        trigger_payload = {
            "id": str(exc.id),
            "route": self.route
        }
        self.trigger_dispatcher.dispatch(trigger_ref, trigger_payload)

        # We only want to request a pause if this has a parent
        if liveaction_db.context.get("parent"):

            # Get the root liveaction and request that it pauses
            root_liveaction = action_service.get_root_liveaction(liveaction_db)
            action_service.request_pause(
                root_liveaction,
                self.context.get('user', None)
            )

        result = {
            "schema": self.schema,
            "roles": self.roles_param,
            "users": self.users_param,
            "route": self.route,
            "ttl": self.ttl
        }
        return (LIVEACTION_STATUS_PENDING, result, None)

 def _submit_request(self):
     context = {'user': USERNAME}
     parameters = {'hosts': 'localhost', 'cmd': 'uname -a'}
     request = LiveActionDB(action=ACTION_REF, context=context, parameters=parameters)
     request, _ = action_service.request(request)
     execution = action_db.get_liveaction_by_id(str(request.id))
     return request, execution

 def test_runner_info(self):
     testworker = worker.Worker(None)
     live_action_db = self._get_execution_db_model()
     testworker.execute_action(live_action_db)
     updated_live_action_db = get_liveaction_by_id(live_action_db.id)
     self.assertEqual(updated_live_action_db.status,
                      action_constants.LIVEACTION_STATUS_RUNNING)
     self.assertTrue(updated_live_action_db.runner_info, 'runner_info should have value.')

 def test_basic_execution_fail(self):
     testworker = worker.Worker(None)
     live_action_db = self._get_execution_db_model(
         status=action_constants.LIVEACTION_STATUS_FAILED)
     testworker.execute_action(live_action_db)
     updated_live_action_db = get_liveaction_by_id(live_action_db.id)
     self.assertEqual(updated_live_action_db.status,
                      action_constants.LIVEACTION_STATUS_FAILED)

    def test_execute(self):
        liveaction_db = self._create_liveaction_db()
        self._process_request(liveaction_db)

        scheduled_liveaction_db = action_db.get_liveaction_by_id(liveaction_db.id)
        scheduled_liveaction_db = self._wait_on_status(
            scheduled_liveaction_db,
            action_constants.LIVEACTION_STATUS_SCHEDULED
        )
        self.assertDictEqual(scheduled_liveaction_db.runner_info, {})

        self.dispatcher._queue_consumer._process_message(scheduled_liveaction_db)
        dispatched_liveaction_db = action_db.get_liveaction_by_id(liveaction_db.id)
        self.assertGreater(len(list(dispatched_liveaction_db.runner_info.keys())), 0)
        self.assertEqual(
            dispatched_liveaction_db.status,
            action_constants.LIVEACTION_STATUS_RUNNING
        )

    def process(self, liveaction):
        """Dispatches the LiveAction to appropriate action runner.

        LiveAction in statuses other than "scheduled" are ignored. If
        LiveAction is already canceled and result is empty, the LiveAction
        is updated with a generic exception message.

        :param liveaction: Scheduled action execution request.
        :type liveaction: ``st2common.models.db.liveaction.LiveActionDB``

        :rtype: ``dict``
        """

        if liveaction.status == action_constants.LIVEACTION_STATUS_CANCELED:
            LOG.info('%s is not executing %s (id=%s) with "%s" status.',
                     self.__class__.__name__, type(liveaction), liveaction.id, liveaction.status)
            if not liveaction.result:
                updated_liveaction = action_utils.update_liveaction_status(
                    status=liveaction.status,
                    result={'message': 'Action execution canceled by user.'},
                    liveaction_id=liveaction.id)
                executions.update_execution(updated_liveaction)
            return

        if liveaction.status != action_constants.LIVEACTION_STATUS_SCHEDULED:
            LOG.info('%s is not executing %s (id=%s) with "%s" status.',
                     self.__class__.__name__, type(liveaction), liveaction.id, liveaction.status)
            return

        try:
            liveaction_db = action_utils.get_liveaction_by_id(liveaction.id)
        except StackStormDBObjectNotFoundError:
            LOG.exception('Failed to find liveaction %s in the database.', liveaction.id)
            raise

        # stamp liveaction with process_info
        runner_info = system_info.get_process_info()

        # Update liveaction status to "running"
        liveaction_db = action_utils.update_liveaction_status(
            status=action_constants.LIVEACTION_STATUS_RUNNING,
            runner_info=runner_info,
            liveaction_id=liveaction_db.id)

        action_execution_db = executions.update_execution(liveaction_db)

        # Launch action
        extra = {'action_execution_db': action_execution_db, 'liveaction_db': liveaction_db}
        LOG.audit('Launching action execution.', extra=extra)

        # the extra field will not be shown in non-audit logs so temporarily log at info.
        LOG.info('Dispatched {~}action_execution: %s / {~}live_action: %s with "%s" status.',
                 action_execution_db.id, liveaction_db.id, liveaction.status)

        return self._run_action(liveaction_db)

    def process(self, request):
        """Schedules the LiveAction and publishes the request
        to the appropriate action runner(s).

        LiveAction in statuses other than "requested" are ignored.

        :param request: Action execution request.
        :type request: ``st2common.models.db.liveaction.LiveActionDB``
        """

        if request.status != action_constants.LIVEACTION_STATUS_REQUESTED:
            LOG.info(
                '%s is ignoring %s (id=%s) with "%s" status.',
                self.__class__.__name__,
                type(request),
                request.id,
                request.status,
            )
            return

        try:
            liveaction_db = action_utils.get_liveaction_by_id(request.id)
        except StackStormDBObjectNotFoundError:
            LOG.exception("Failed to find liveaction %s in the database.", request.id)
            raise

        # Apply policies defined for the action.
        liveaction_db = self._apply_pre_run_policies(liveaction_db=liveaction_db)

        # Exit if the status of the request is no longer runnable.
        # The status could have be changed by one of the policies.
        if liveaction_db.status not in [
            action_constants.LIVEACTION_STATUS_REQUESTED,
            action_constants.LIVEACTION_STATUS_SCHEDULED,
        ]:
            LOG.info(
                '%s is ignoring %s (id=%s) with "%s" status after policies are applied.',
                self.__class__.__name__,
                type(request),
                request.id,
                liveaction_db.status,
            )
            return

        # Update liveaction status to "scheduled".
        if liveaction_db.status == action_constants.LIVEACTION_STATUS_REQUESTED:
            liveaction_db = action_service.update_status(
                liveaction_db, action_constants.LIVEACTION_STATUS_SCHEDULED, publish=False
            )

        # Publish the "scheduled" status here manually. Otherwise, there could be a
        # race condition with the update of the action_execution_db if the execution
        # of the liveaction completes first.
        LiveAction.publish_status(liveaction_db)

    def process(self, liveaction):
        """Dispatches the LiveAction to appropriate action runner.

        LiveAction in statuses other than "scheduled" and "canceling" are ignored. If
        LiveAction is already canceled and result is empty, the LiveAction
        is updated with a generic exception message.

        :param liveaction: Action execution request.
        :type liveaction: ``st2common.models.db.liveaction.LiveActionDB``

        :rtype: ``dict``
        """

        if liveaction.status == action_constants.LIVEACTION_STATUS_CANCELED:
            LOG.info(
                '%s is not executing %s (id=%s) with "%s" status.',
                self.__class__.__name__,
                type(liveaction),
                liveaction.id,
                liveaction.status,
            )
            if not liveaction.result:
                updated_liveaction = action_utils.update_liveaction_status(
                    status=liveaction.status,
                    result={"message": "Action execution canceled by user."},
                    liveaction_id=liveaction.id,
                )
                executions.update_execution(updated_liveaction)
            return

        if liveaction.status not in [
            action_constants.LIVEACTION_STATUS_SCHEDULED,
            action_constants.LIVEACTION_STATUS_CANCELING,
        ]:
            LOG.info(
                '%s is not dispatching %s (id=%s) with "%s" status.',
                self.__class__.__name__,
                type(liveaction),
                liveaction.id,
                liveaction.status,
            )
            return

        try:
            liveaction_db = action_utils.get_liveaction_by_id(liveaction.id)
        except StackStormDBObjectNotFoundError:
            LOG.exception("Failed to find liveaction %s in the database.", liveaction.id)
            raise

        return (
            self._run_action(liveaction_db)
            if liveaction.status == action_constants.LIVEACTION_STATUS_SCHEDULED
            else self._cancel_action(liveaction_db)
        )

 def test_failed_execution_handling(self):
     testworker = worker.Worker(None)
     live_action_db = self._get_execution_db_model()
     try:
         testworker.execute_action(live_action_db)
         self.assertTrue(False, 'Exception expected.')
     except Exception:
         self.assertTrue(True)
     updated_live_action_db = get_liveaction_by_id(live_action_db.id)
     self.assertEqual(updated_live_action_db.status,
                      action_constants.LIVEACTION_STATUS_FAILED)

    def _handle_execution(self, execution_queue_item_db):
        liveaction_id = str(execution_queue_item_db.liveaction_id)
        queue_item_id = str(execution_queue_item_db.id)

        extra = {
            'liveaction_id': liveaction_id,
            'queue_item_id': queue_item_id
        }

        LOG.info('Scheduling liveaction: %s (queue_item_id=%s)', liveaction_id,
                 queue_item_id, extra=extra)

        try:
            liveaction_db = action_utils.get_liveaction_by_id(liveaction_id)
        except StackStormDBObjectNotFoundError:
            LOG.exception('Failed to find liveaction %s in the database (queue_item_id=%s).',
                          liveaction_id, queue_item_id, extra=extra)
            ActionExecutionSchedulingQueue.delete(execution_queue_item_db)
            raise

        # Identify if the action has policies that require locking.
        action_has_policies_require_lock = policy_service.has_policies(
            liveaction_db,
            policy_types=policy_constants.POLICY_TYPES_REQUIRING_LOCK
        )

        # Acquire a distributed lock if the referenced action has specific policies attached.
        if action_has_policies_require_lock:
            # Warn users that the coordination service is not configured.
            if not coordination_service.configured():
                LOG.warn(
                    'Coordination backend is not configured. '
                    'Policy enforcement is best effort.'
                )

            # Acquire a distributed lock before querying the database to make sure that only one
            # scheduler is scheduling execution for this action. Even if the coordination service
            # is not configured, the fake driver using zake or the file driver can still acquire
            # a lock for the local process or server respectively.
            lock_uid = liveaction_db.action
            LOG.debug('%s is attempting to acquire lock "%s".', self.__class__.__name__, lock_uid)
            lock = self._coordinator.get_lock(lock_uid)

            try:
                if lock.acquire(blocking=False):
                    self._regulate_and_schedule(liveaction_db, execution_queue_item_db)
                else:
                    self._delay(liveaction_db, execution_queue_item_db)
            finally:
                lock.release()
        else:
            # Otherwise if there is no policy, then schedule away.
            self._schedule(liveaction_db, execution_queue_item_db)