def _dtw_schedule_immediately(self, ctx):

        for d in dtw.get_unscheduled_delay_tolerant_workload():
            LOG.debug("Processing delay tolerant workload: %s" % d)

            # Setup admin context before schedule triggers.
            ctx = security.create_context(d.trust_id, d.project_id)

            auth_ctx.set_ctx(ctx)

            LOG.debug("Delay tolerant workload security context: %s" % ctx)

            try:
                # execute the workload

                db_api_v2.update_delay_tolerant_workload(
                    d.name,
                    {'executed': True}
                )

                rpc.get_engine_client().start_workflow(
                    d.workflow.name,
                    d.workflow_input,
                    description="DTW Workflow execution created.",
                    **d.workflow_params
                )
            except Exception:
                # Log and continue to next cron trigger.
                LOG.exception(
                    "Failed to process delay tolerant workload %s" % str(d))
            finally:
                auth_ctx.set_ctx(None)

    def process_cron_triggers_v2(self, ctx):
        for t in triggers.get_next_cron_triggers():
            LOG.debug("Processing cron trigger: %s" % t)

            # Setup admin context before schedule triggers.
            ctx = security.create_context(t.trust_id, t.project_id)

            auth_ctx.set_ctx(ctx)

            LOG.debug("Cron trigger security context: %s" % ctx)

            try:
                rpc.get_engine_client().start_workflow(
                    t.workflow.name,
                    t.workflow_input,
                    description="Workflow execution created by cron trigger.",
                    **t.workflow_params
                )
            finally:
                if t.remaining_executions is not None and t.remaining_executions > 0:
                    t.remaining_executions -= 1
                if t.remaining_executions == 0:
                    db_api_v2.delete_cron_trigger(t.name)
                else:  # if remaining execution = None or > 0
                    next_time = triggers.get_next_execution_time(t.pattern, t.next_execution_time)

                    db_api_v2.update_cron_trigger(
                        t.name, {"next_execution_time": next_time, "remaining_executions": t.remaining_executions}
                    )

                    auth_ctx.set_ctx(None)

    def _invoke_calls(delayed_calls):
        """Invokes prepared delayed calls.

        :param delayed_calls: Prepared delayed calls represented as tuples
        (target_auth_context, target_method, method_args).
        """

        ctx_serializer = context.RpcContextSerializer()

        for (target_auth_context, target_method, method_args) in delayed_calls:
            try:
                # Set the correct context for the method.
                ctx_serializer.deserialize_context(target_auth_context)

                # Invoke the method.
                target_method(**method_args)
            except Exception as e:
                LOG.exception(
                    "Delayed call failed, method: %s, exception: %s",
                    target_method,
                    e
                )
            finally:
                # Remove context.
                context.set_ctx(None)

        def run_delayed_task(context):
            """Runs the delayed task. Performs all the steps required to setup
            a task to run which are not already done. This is mostly code
            copied over from convey_task_result.

            :param context Mistral authentication context inherited from a
                caller thread.
            """
            auth_context.set_ctx(context)

            db_api.start_tx()

            try:
                execution_id = task['execution_id']
                execution = db_api.execution_get(execution_id)

                # Change state from DELAYED to RUNNING.

                WORKFLOW_TRACE.info("Task '%s' [%s -> %s]"
                                    % (task['name'],
                                       task['state'], states.RUNNING))
                executables = data_flow.prepare_tasks([task],
                                                      outbound_context,
                                                      workbook)
                db_api.commit_tx()
            finally:
                db_api.end_tx()

            if states.is_stopped_or_finished(execution['state']):
                return

            for task_id, action_name, action_params in executables:
                self._run_task(task_id, action_name, action_params)

    def _start_workflow(self, triggers, payload, metadata):
        """Start workflows defined in event triggers."""
        for t in triggers:
            LOG.info('Start to process event trigger: %s', t['id'])

            workflow_params = t.get('workflow_params', {})
            workflow_params.update(
                {'event_payload': payload, 'event_metadata': metadata}
            )

            # Setup context before schedule triggers.
            ctx = security.create_context(t['trust_id'], t['project_id'])
            auth_ctx.set_ctx(ctx)

            try:
                self.engine_client.start_workflow(
                    t['workflow_id'],
                    t['workflow_input'],
                    description="Workflow execution created by event "
                                "trigger %s." % t['id'],
                    **workflow_params
                )
            except Exception as e:
                LOG.exception("Failed to process event trigger %s, "
                              "error: %s", t['id'], str(e))
            finally:
                auth_ctx.set_ctx(None)

def run_execution_expiration_policy(self, ctx):
    LOG.debug("Starting expiration policy task.")

    older_than = CONF.execution_expiration_policy.older_than
    exp_time = (datetime.datetime.now()
                - datetime.timedelta(minutes=older_than))

    with db_api.transaction():
        # TODO(gpaz): In the future should use generic method with
        # filters params and not specific method that filter by time.
        for execution in db_api.get_expired_executions(exp_time):
            try:
                # Setup project_id for _secure_query delete execution.
                ctx = auth_ctx.MistralContext(
                    user_id=None,
                    project_id=execution.project_id,
                    auth_token=None,
                    is_admin=True
                )
                auth_ctx.set_ctx(ctx)

                LOG.debug(
                    'DELETE execution id : %s from date : %s '
                    'according to expiration policy',
                    execution.id,
                    execution.updated_at
                )
                db_api.delete_workflow_execution(execution.id)
            except Exception as e:
                msg = ("Failed to delete [execution_id=%s]\n %s"
                       % (execution.id, traceback.format_exc(e)))
                LOG.warning(msg)
            finally:
                auth_ctx.set_ctx(None)

    def test_workflow_definition_public(self):
        # Create a workflow(scope=public) as under one project
        # then make sure it's visible for other projects.
        created0 = db_api.create_workflow_definition(WF_DEFINITIONS[0])

        fetched = db_api.get_workflow_definitions()

        self.assertEqual(1, len(fetched))
        self.assertEqual(created0, fetched[0])

        # Assert that the project_id stored is actually the context's
        # project_id not the one given.
        self.assertEqual(created0.project_id, auth_context.ctx().project_id)
        self.assertNotEqual(
            WF_DEFINITIONS[0]['project_id'],
            auth_context.ctx().project_id
        )

        # Create a new user.
        auth_context.set_ctx(test_base.get_context(default=False))

        fetched = db_api.get_workflow_definitions()

        self.assertEqual(1, len(fetched))
        self.assertEqual(created0, fetched[0])
        self.assertEqual('public', created0.scope)

    def process_cron_triggers_v2(self, ctx):
        for t in triggers.get_next_cron_triggers():
            LOG.debug("Processing cron trigger: %s" % t)

            # Setup admin context before schedule triggers.
            ctx = security.create_context(t.trust_id, t.project_id)

            auth_ctx.set_ctx(ctx)

            LOG.debug("Cron trigger security context: %s" % ctx)

            try:
                # Try to advance the cron trigger next_execution_time and
                # remaining_executions if relevant.
                modified = advance_cron_trigger(t)

                # If cron trigger was not already modified by another engine.
                if modified:
                    LOG.debug(
                        "Starting workflow '%s' by cron trigger '%s'",
                        t.workflow.name, t.name
                    )

                    rpc.get_engine_client().start_workflow(
                        t.workflow.name,
                        t.workflow_input,
                        description="Workflow execution created "
                                    "by cron trigger.",
                        **t.workflow_params
                    )
            except Exception:
                # Log and continue to next cron trigger.
                LOG.exception("Failed to process cron trigger %s" % str(t))
            finally:
                auth_ctx.set_ctx(None)

    def test_scheduler_call_target_method_with_correct_auth(self, method):
        method.side_effect = self.target_check_context_method

        default_context = base.get_context(default=True)
        auth_context.set_ctx(default_context)
        default_project_id = (
            default_context.project_id
        )

        scheduler.schedule_call(
            None,
            TARGET_METHOD_PATH,
            DELAY,
            **{'expected_project_id': default_project_id}
        )

        second_context = base.get_context(default=False)
        auth_context.set_ctx(second_context)
        second_project_id = (
            second_context.project_id
        )

        scheduler.schedule_call(
            None,
            TARGET_METHOD_PATH,
            DELAY,
            **{'expected_project_id': second_project_id}
        )

        self.assertNotEqual(default_project_id, second_project_id)

        for _ in range(2):
            self.assertTrue(self.queue.get())

def _delete(executions):
    for execution in executions:
        try:
            # Setup project_id for _secure_query delete execution.
            # TODO(tuan_luong): Manipulation with auth_ctx should be
            # out of db transaction scope.
            ctx = auth_ctx.MistralContext(
                user=None,
                tenant=execution.project_id,
                auth_token=None,
                is_admin=True
            )
            auth_ctx.set_ctx(ctx)

            LOG.debug(
                'Delete execution id : %s from date : %s '
                'according to expiration policy',
                execution.id,
                execution.updated_at
            )
            db_api.delete_workflow_execution(execution.id)
        except Exception as e:
            msg = ("Failed to delete [execution_id=%s]\n %s"
                   % (execution.id, traceback.format_exc(e)))
            LOG.warning(msg)
        finally:
            auth_ctx.set_ctx(None)

    def test_workbook_public(self):
        # create a workbook(scope=public) as under one project
        # then make sure it's visible for other projects.
        created0 = db_api.workbook_create(WORKBOOKS[0])

        fetched = db_api.workbooks_get_all()

        self.assertEqual(1, len(fetched))
        self.assertDictEqual(created0, fetched[0])

        # assert that the project_id stored is actually the context's
        # project_id not the one given.
        self.assertEqual(created0['project_id'], auth_context.ctx().project_id)
        self.assertNotEqual(WORKBOOKS[0]['project_id'],
                            auth_context.ctx().project_id)

        # create a new user.
        ctx = auth_context.MistralContext(user_id='9-0-44-5',
                                          project_id='99-88-33',
                                          user_name='test-user',
                                          project_name='test-another',
                                          is_admin=False)
        auth_context.set_ctx(ctx)

        fetched = db_api.workbooks_get_all()
        self.assertEqual(1, len(fetched))
        self.assertDictEqual(created0, fetched[0])
        self.assertEqual('public', created0['scope'])

    def _dtw_last_minute_scheduling(self, ctx):

        for d in dtw.get_unscheduled_delay_tolerant_workload():
            LOG.debug("Processing delay tolerant workload: %s" % d)

            # Setup admin context before schedule triggers.
            ctx = security.create_context(d.trust_id, d.project_id)

            auth_ctx.set_ctx(ctx)

            LOG.debug("Delay tolerant workload security context: %s" % ctx)

            # calculate last time for running this - deadline less the
            # duration of the work
            # TODO(murp): check the status of the security context on this
            # TODO(murp): convert job_duration to timedelta
            start_time = d.deadline - datetime.timedelta(seconds=d.job_duration)
            
            triggers.create_cron_trigger(d.name, d.workflow_name, 
                                         d.workflow_input,
                                         workflow_params=d.workflow_params, 
                                         count=1, 
                                         first_time=start_time,
                                         start_time=start_time, 
                                         workflow_id=d.workflow_id)

def _loop():
    global _stopped

    # This is an administrative thread so we need to set an admin
    # security context.
    auth_ctx.set_ctx(
        auth_ctx.MistralContext(
            user=None,
            tenant=None,
            auth_token=None,
            is_admin=True
        )
    )

    while not _stopped:
        try:
            handle_expired_actions()
        except Exception:
            LOG.exception(
                'Action execution checker iteration failed'
                ' due to unexpected exception.'
            )

            # For some mysterious reason (probably eventlet related)
            # the exception is not cleared from the context automatically.
            # This results in subsequent log.warning calls to show invalid
            # info.
            if sys.version_info < (3,):
                sys.exc_clear()

        eventlet.sleep(CONF.action_heartbeat.check_interval)

    def _clean_db(self):
        contexts = [
            get_context(default=False),
            get_context(default=True)
        ]

        for ctx in contexts:
            auth_context.set_ctx(ctx)

            with mock.patch('mistral.services.security.get_project_id',
                            new=mock.MagicMock(return_value=ctx.project_id)):
                with db_api.transaction():
                    db_api.delete_event_triggers()
                    db_api.delete_cron_triggers()
                    db_api.delete_workflow_executions()
                    db_api.delete_task_executions()
                    db_api.delete_action_executions()
                    db_api.delete_workbooks()
                    db_api.delete_workflow_definitions()
                    db_api.delete_environments()
                    db_api.delete_resource_members()

        sqlite_lock.cleanup()

        if not cfg.CONF.database.connection.startswith('sqlite'):
            db_sa_base.get_engine().dispose()

    def test_workflow_definition_public(self):
        # Create a workflow(scope=public) as under one project
        # then make sure it's visible for other projects.
        created0 = db_api.create_workflow_definition(WF_DEFINITIONS[0])

        fetched = db_api.get_workflow_definitions()

        self.assertEqual(1, len(fetched))
        self.assertEqual(created0, fetched[0])

        # Assert that the project_id stored is actually the context's
        # project_id not the one given.
        self.assertEqual(created0.project_id, auth_context.ctx().project_id)
        self.assertNotEqual(
            WF_DEFINITIONS[0]['project_id'],
            auth_context.ctx().project_id
        )

        # Create a new user.
        ctx = auth_context.MistralContext(
            user_id='9-0-44-5',
            project_id='99-88-33',
            user_name='test-user',
            project_name='test-another',
            is_admin=False
        )

        auth_context.set_ctx(ctx)

        fetched = db_api.get_workflow_definitions()

        self.assertEqual(1, len(fetched))
        self.assertEqual(created0, fetched[0])
        self.assertEqual('public', created0.scope)

    def _set_auth_ctx(ctx):
        if not isinstance(ctx, dict):
            return

        context = auth_ctx.MistralContext.from_dict(ctx)
        auth_ctx.set_ctx(context)

        return context

    def _set_auth_ctx(ctx):
        if not isinstance(ctx, dict):
            return

        context = auth_context.MistralContext(**ctx)
        auth_context.set_ctx(context)

        return context

def process_cron_triggers_v2(self, ctx):
    LOG.debug("Processing cron triggers...")

    for trigger in triggers.get_next_cron_triggers():
        LOG.debug("Processing cron trigger: %s", trigger)

        try:
            # Setup admin context before schedule triggers.
            ctx = security.create_context(
                trigger.trust_id,
                trigger.project_id
            )

            auth_ctx.set_ctx(ctx)

            LOG.debug("Cron trigger security context: %s", ctx)

            # Try to advance the cron trigger next_execution_time and
            # remaining_executions if relevant.
            modified = advance_cron_trigger(trigger)

            # If cron trigger was not already modified by another engine.
            if modified:
                LOG.debug(
                    "Starting workflow '%s' by cron trigger '%s'",
                    trigger.workflow.name,
                    trigger.name
                )

                description = {
                    "description": (
                        "Workflow execution created by cron"
                        " trigger '(%s)'." % trigger.id
                    ),
                    "triggered_by": {
                        "type": "cron_trigger",
                        "id": trigger.id,
                        "name": trigger.name,
                    }
                }

                rpc.get_engine_client().start_workflow(
                    trigger.workflow.name,
                    trigger.workflow.namespace,
                    None,
                    trigger.workflow_input,
                    description=json.dumps(description),
                    **trigger.workflow_params
                )
        except Exception:
            # Log and continue to next cron trigger.
            LOG.exception(
                "Failed to process cron trigger %s",
                str(trigger)
            )
        finally:
            auth_ctx.set_ctx(None)

def _switch_context(project_id, is_admin):
    _ctx = ctx.MistralContext(
        user_id=None,
        project_id=project_id,
        auth_token=None,
        is_admin=is_admin
    )

    ctx.set_ctx(_ctx)

    def tearDown(self):
        """Restores the size limit config to default."""
        super(ExpirationPolicyTest, self).tearDown()

        cfg.CONF.set_default('auth_enable', False, group='pecan')

        ctx.set_ctx(None)

        _set_expiration_policy_config(None, None)