def test_no_timestamp_doesnt_delete_things(self):
        now = date_utils.get_datetime_utc_now()
        exec_model = copy.deepcopy(self.models['executions']['execution1.yaml'])
        exec_model['start_timestamp'] = now - timedelta(days=15)
        exec_model['end_timestamp'] = now - timedelta(days=14)
        exec_model['status'] = action_constants.LIVEACTION_STATUS_SUCCEEDED
        exec_model['id'] = bson.ObjectId()
        ActionExecution.add_or_update(exec_model)

        # Insert corresponding stdout and stderr db mock models
        self._insert_mock_stdout_and_stderr_objects_for_execution(exec_model['id'], count=3)

        execs = ActionExecution.get_all()
        self.assertEqual(len(execs), 1)
        stdout_dbs = ActionExecutionOutput.query(output_type='stdout')
        self.assertEqual(len(stdout_dbs), 3)
        stderr_dbs = ActionExecutionOutput.query(output_type='stderr')
        self.assertEqual(len(stderr_dbs), 3)

        expected_msg = 'Specify a valid timestamp'
        self.assertRaisesRegexp(ValueError, expected_msg, purge_executions,
                                logger=LOG, timestamp=None)
        execs = ActionExecution.get_all()
        self.assertEqual(len(execs), 1)
        stdout_dbs = ActionExecutionOutput.query(output_type='stdout')
        self.assertEqual(len(stdout_dbs), 3)
        stderr_dbs = ActionExecutionOutput.query(output_type='stderr')
        self.assertEqual(len(stderr_dbs), 3)

def _purge_action_models(execution_db):
    liveaction_id = execution_db.liveaction['id']

    if not liveaction_id:
        print('Invalid LiveAction id. Skipping delete: %s', execution_db)

    liveaction_db = None
    try:
        liveaction_db = LiveAction.get_by_id(liveaction_id)
    except:
        print('LiveAction with id: %s not found. Skipping delete.', liveaction_id)
    else:
        global DELETED_COUNT
        DELETED_COUNT += 1

    try:
        ActionExecution.delete(execution_db)
    except Exception as e:
        print('Exception deleting Execution model: %s, exception: %s',
              execution_db, str(e))
    else:
        try:
            LiveAction.delete(liveaction_db)
        except Exception as e:
            print('Zombie LiveAction left in db: %s. Exception: %s',
                  liveaction_db, str(e))

    def _get_runner(self, runnertype_db, action_db, liveaction_db):
        runner = get_runner(runnertype_db.runner_module)

        resolved_entry_point = self._get_entry_point_abs_path(action_db.pack,
                                                              action_db.entry_point)

        runner.runner_type_db = runnertype_db
        runner.container_service = RunnerContainerService()
        runner.action = action_db
        runner.action_name = action_db.name
        runner.liveaction = liveaction_db
        runner.liveaction_id = str(liveaction_db.id)
        runner.execution = ActionExecution.get(liveaction__id=runner.liveaction_id)
        runner.execution_id = str(runner.execution.id)
        runner.entry_point = resolved_entry_point
        runner.context = getattr(liveaction_db, 'context', dict())
        runner.callback = getattr(liveaction_db, 'callback', dict())
        runner.libs_dir_path = self._get_action_libs_abs_path(action_db.pack,
                                                              action_db.entry_point)

        # For re-run, get the ActionExecutionDB in which the re-run is based on.
        rerun_ref_id = runner.context.get('re-run', {}).get('ref')
        runner.rerun_ex_ref = ActionExecution.get(id=rerun_ref_id) if rerun_ref_id else None

        return runner

def update_execution(liveaction_db, publish=True):
    execution = ActionExecution.get(liveaction__id=str(liveaction_db.id))
    decomposed = _decompose_liveaction(liveaction_db)
    for k, v in six.iteritems(decomposed):
        setattr(execution, k, v)
    execution = ActionExecution.add_or_update(execution, publish=publish)
    return execution

    def test_chain_cancel_cascade_to_subworkflow(self):
        # A temp file is created during test setup. Ensure the temp file exists.
        # The test action chain will stall until this file is deleted. This gives
        # the unit test a moment to run any test related logic.
        path = self.temp_file_path
        self.assertTrue(os.path.exists(path))

        action = TEST_PACK + '.' + 'test_cancel_with_subworkflow'
        params = {'tempfile': path, 'message': 'foobar'}
        liveaction = LiveActionDB(action=action, parameters=params)
        liveaction, execution = action_service.request(liveaction)
        liveaction = LiveAction.get_by_id(str(liveaction.id))

        # Wait until the liveaction is running.
        liveaction = self._wait_on_status(liveaction, action_constants.LIVEACTION_STATUS_RUNNING)

        # Wait for subworkflow to register.
        execution = self._wait_for_children(execution)
        self.assertEqual(len(execution.children), 1)

        # Wait until the subworkflow is running.
        task1_exec = ActionExecution.get_by_id(execution.children[0])
        task1_live = LiveAction.get_by_id(task1_exec.liveaction['id'])
        task1_live = self._wait_on_status(task1_live, action_constants.LIVEACTION_STATUS_RUNNING)

        # Request action chain to cancel.
        liveaction, execution = action_service.request_cancellation(liveaction, USERNAME)

        # Wait until the liveaction is canceling.
        liveaction = self._wait_on_status(liveaction, action_constants.LIVEACTION_STATUS_CANCELING)
        self.assertEqual(len(execution.children), 1)

        # Wait until the subworkflow is canceling.
        task1_exec = ActionExecution.get_by_id(execution.children[0])
        task1_live = LiveAction.get_by_id(task1_exec.liveaction['id'])
        task1_live = self._wait_on_status(task1_live, action_constants.LIVEACTION_STATUS_CANCELING)

        # Delete the temporary file that the action chain is waiting on.
        os.remove(path)
        self.assertFalse(os.path.exists(path))

        # Wait until the liveaction is canceled.
        liveaction = self._wait_on_status(liveaction, action_constants.LIVEACTION_STATUS_CANCELED)
        self.assertEqual(len(execution.children), 1)

        # Wait until the subworkflow is canceled.
        task1_exec = ActionExecution.get_by_id(execution.children[0])
        task1_live = LiveAction.get_by_id(task1_exec.liveaction['id'])
        task1_live = self._wait_on_status(task1_live, action_constants.LIVEACTION_STATUS_CANCELED)

        # Wait for non-blocking threads to complete. Ensure runner is not running.
        MockLiveActionPublisherNonBlocking.wait_all()

        # Check liveaction result.
        self.assertIn('tasks', liveaction.result)
        self.assertEqual(len(liveaction.result['tasks']), 1)

        subworkflow = liveaction.result['tasks'][0]
        self.assertEqual(len(subworkflow['result']['tasks']), 1)
        self.assertEqual(subworkflow['state'], action_constants.LIVEACTION_STATUS_CANCELED)

    def _has_active_tasks(self, liveaction_db, mistral_wf_state, mistral_tasks):
        # Identify if there are any active tasks in Mistral.
        active_mistral_tasks = len([t for t in mistral_tasks if t['state'] in ACTIVE_STATES]) > 0

        active_st2_tasks = False
        execution = ActionExecution.get(liveaction__id=str(liveaction_db.id))

        for child_exec_id in execution.children:
            child_exec = ActionExecution.get(id=child_exec_id)

            # Catch exception where a child is requested twice due to st2mistral retrying
            # from a st2 API connection failure. The first child will be stuck in requested
            # while the mistral workflow is already completed.
            if (mistral_wf_state in DONE_STATES and
                    child_exec.status == action_constants.LIVEACTION_STATUS_REQUESTED):
                continue

            if (child_exec.status not in action_constants.LIVEACTION_COMPLETED_STATES and
                    child_exec.status != action_constants.LIVEACTION_STATUS_PAUSED):
                active_st2_tasks = True
                break

        if active_mistral_tasks:
            LOG.info('There are active mistral tasks for %s.', str(liveaction_db.id))

        if active_st2_tasks:
            LOG.info('There are active st2 tasks for %s.', str(liveaction_db.id))

        return active_mistral_tasks or active_st2_tasks

def _purge_models(execution_db):
    liveaction_id = execution_db.liveaction.get("id", None)

    if not liveaction_id:
        LOG.error("Invalid LiveAction id. Skipping delete: %s", execution_db)

    liveaction_db = None
    try:
        liveaction_db = LiveAction.get_by_id(liveaction_id)
    except:
        LOG.exception("LiveAction with id: %s not found. Skipping delete.", liveaction_id)
    else:
        global DELETED_COUNT
        DELETED_COUNT += 1

    try:
        ActionExecution.delete(execution_db)
    except:
        LOG.exception("Exception deleting Execution model: %s", execution_db)
    else:
        if liveaction_db:
            try:
                LiveAction.delete(liveaction_db)
            except:
                LOG.exception("Zombie LiveAction left in db: %s.", liveaction_db)

    def test_liveaction_gets_deleted(self):
        now = date_utils.get_datetime_utc_now()
        start_ts = now - timedelta(days=15)
        end_ts = now - timedelta(days=14)

        liveaction_model = copy.deepcopy(self.models['liveactions']['liveaction4.yaml'])
        liveaction_model['start_timestamp'] = start_ts
        liveaction_model['end_timestamp'] = end_ts
        liveaction_model['status'] = action_constants.LIVEACTION_STATUS_SUCCEEDED
        liveaction = LiveAction.add_or_update(liveaction_model)

        # Write one execution before cut-off threshold
        exec_model = copy.deepcopy(self.models['executions']['execution1.yaml'])
        exec_model['start_timestamp'] = start_ts
        exec_model['end_timestamp'] = end_ts
        exec_model['status'] = action_constants.LIVEACTION_STATUS_SUCCEEDED
        exec_model['id'] = bson.ObjectId()
        exec_model['liveaction']['id'] = str(liveaction.id)
        ActionExecution.add_or_update(exec_model)

        liveactions = LiveAction.get_all()
        executions = ActionExecution.get_all()
        self.assertEqual(len(liveactions), 1)
        self.assertEqual(len(executions), 1)

        purge_executions(logger=LOG, timestamp=now - timedelta(days=10))

        liveactions = LiveAction.get_all()
        executions = ActionExecution.get_all()
        self.assertEqual(len(executions), 0)
        self.assertEqual(len(liveactions), 0)

def get_descendants(actionexecution_id, descendant_depth=-1, result_fmt=None):
    """
    Returns all descendant executions upto the specified descendant_depth for
    the supplied actionexecution_id.
    """
    descendants = DESCENDANT_VIEWS.get(result_fmt, DFSDescendantView)()
    children = ActionExecution.query(parent=actionexecution_id,
                                     **{'order_by': ['start_timestamp']})
    LOG.debug('Found %s children for id %s.', len(children), actionexecution_id)
    current_level = [(child, 1) for child in children]

    while current_level:
        parent, level = current_level.pop(0)
        parent_id = str(parent.id)
        descendants.add(parent)
        if not parent.children:
            continue
        if level != -1 and level == descendant_depth:
            continue
        children = ActionExecution.query(parent=parent_id, **{'order_by': ['start_timestamp']})
        LOG.debug('Found %s children for id %s.', len(children), parent_id)
        # prepend for DFS
        for idx in range(len(children)):
            current_level.insert(idx, (children[idx], level + 1))
    return descendants.result

def create_execution_object(liveaction, action_db=None, runnertype_db=None, publish=True):
    if not action_db:
        action_db = action_utils.get_action_by_ref(liveaction.action)

    if not runnertype_db:
        runnertype_db = RunnerType.get_by_name(action_db.runner_type['name'])

    attrs = {
        'action': vars(ActionAPI.from_model(action_db)),
        'parameters': liveaction['parameters'],
        'runner': vars(RunnerTypeAPI.from_model(runnertype_db))
    }
    attrs.update(_decompose_liveaction(liveaction))

    if 'rule' in liveaction.context:
        rule = reference.get_model_from_ref(Rule, liveaction.context.get('rule', {}))
        attrs['rule'] = vars(RuleAPI.from_model(rule))

    if 'trigger_instance' in liveaction.context:
        trigger_instance_id = liveaction.context.get('trigger_instance', {})
        trigger_instance_id = trigger_instance_id.get('id', None)
        trigger_instance = TriggerInstance.get_by_id(trigger_instance_id)
        trigger = reference.get_model_by_resource_ref(db_api=Trigger,
                                                      ref=trigger_instance.trigger)
        trigger_type = reference.get_model_by_resource_ref(db_api=TriggerType,
                                                           ref=trigger.type)
        trigger_instance = reference.get_model_from_ref(
            TriggerInstance, liveaction.context.get('trigger_instance', {}))
        attrs['trigger_instance'] = vars(TriggerInstanceAPI.from_model(trigger_instance))
        attrs['trigger'] = vars(TriggerAPI.from_model(trigger))
        attrs['trigger_type'] = vars(TriggerTypeAPI.from_model(trigger_type))

    parent = _get_parent_execution(liveaction)
    if parent:
        attrs['parent'] = str(parent.id)

    attrs['log'] = [_create_execution_log_entry(liveaction['status'])]

    # TODO: This object initialization takes 20-30or so ms
    execution = ActionExecutionDB(**attrs)
    # TODO: Do 100% research this is fully safe and unique in distributed setups
    execution.id = ObjectId()
    execution.web_url = _get_web_url_for_execution(str(execution.id))

    # NOTE: User input data is already validate as part of the API request,
    # other data is set by us. Skipping validation here makes operation 10%-30% faster
    execution = ActionExecution.add_or_update(execution, publish=publish, validate=False)

    if parent and str(execution.id) not in parent.children:
        values = {}
        values['push__children'] = str(execution.id)
        ActionExecution.update(parent, **values)

    return execution

    def test_no_timestamp_doesnt_delete_things(self):
        now = date_utils.get_datetime_utc_now()
        exec_model = copy.copy(self.models['executions']['execution1.yaml'])
        exec_model['start_timestamp'] = now - timedelta(days=15)
        exec_model['end_timestamp'] = now - timedelta(days=14)
        exec_model['status'] = action_constants.LIVEACTION_STATUS_SUCCEEDED
        exec_model['id'] = bson.ObjectId()
        ActionExecution.add_or_update(exec_model)

        execs = ActionExecution.get_all()
        self.assertEqual(len(execs), 1)
        purge_executions()
        execs = ActionExecution.get_all()
        self.assertEqual(len(execs), 1)

def update_execution(liveaction_db, publish=True):
    execution = ActionExecution.get(liveaction__id=str(liveaction_db.id))
    decomposed = _decompose_liveaction(liveaction_db)

    kw = {}
    for k, v in six.iteritems(decomposed):
        kw['set__' + k] = v

    if liveaction_db.status != execution.status:
        # Note: If the status changes we store this transition in the "log" attribute of action
        # execution
        kw['push__log'] = _create_execution_log_entry(liveaction_db.status)
    execution = ActionExecution.update(execution, publish=publish, **kw)
    return execution

    def test_crud_complete(self):
        # Create the DB record.
        obj = ActionExecutionAPI(**copy.deepcopy(self.fake_history_workflow))
        ActionExecution.add_or_update(ActionExecutionAPI.to_model(obj))
        model = ActionExecution.get_by_id(obj.id)
        self.assertEqual(str(model.id), obj.id)
        self.assertDictEqual(model.trigger, self.fake_history_workflow['trigger'])
        self.assertDictEqual(model.trigger_type, self.fake_history_workflow['trigger_type'])
        self.assertDictEqual(model.trigger_instance, self.fake_history_workflow['trigger_instance'])
        self.assertDictEqual(model.rule, self.fake_history_workflow['rule'])
        self.assertDictEqual(model.action, self.fake_history_workflow['action'])
        self.assertDictEqual(model.runner, self.fake_history_workflow['runner'])
        doc = copy.deepcopy(self.fake_history_workflow['liveaction'])
        doc['start_timestamp'] = doc['start_timestamp']
        doc['end_timestamp'] = doc['end_timestamp']
        self.assertDictEqual(model.liveaction, doc)
        self.assertIsNone(getattr(model, 'parent', None))
        self.assertListEqual(model.children, self.fake_history_workflow['children'])

        # Update the DB record.
        children = [str(bson.ObjectId()), str(bson.ObjectId())]
        model.children = children
        ActionExecution.add_or_update(model)
        model = ActionExecution.get_by_id(obj.id)
        self.assertListEqual(model.children, children)

        # Delete the DB record.
        ActionExecution.delete(model)
        self.assertRaises(ValueError, ActionExecution.get_by_id, obj.id)

def create_execution_object(liveaction, publish=True):
    action_db = action_utils.get_action_by_ref(liveaction.action)
    runner = RunnerType.get_by_name(action_db.runner_type['name'])

    attrs = {
        'action': vars(ActionAPI.from_model(action_db)),
        'parameters': liveaction['parameters'],
        'runner': vars(RunnerTypeAPI.from_model(runner))
    }
    attrs.update(_decompose_liveaction(liveaction))

    if 'rule' in liveaction.context:
        rule = reference.get_model_from_ref(Rule, liveaction.context.get('rule', {}))
        attrs['rule'] = vars(RuleAPI.from_model(rule))

    if 'trigger_instance' in liveaction.context:
        trigger_instance_id = liveaction.context.get('trigger_instance', {})
        trigger_instance_id = trigger_instance_id.get('id', None)
        trigger_instance = TriggerInstance.get_by_id(trigger_instance_id)
        trigger = reference.get_model_by_resource_ref(db_api=Trigger,
                                                      ref=trigger_instance.trigger)
        trigger_type = reference.get_model_by_resource_ref(db_api=TriggerType,
                                                           ref=trigger.type)
        trigger_instance = reference.get_model_from_ref(
            TriggerInstance, liveaction.context.get('trigger_instance', {}))
        attrs['trigger_instance'] = vars(TriggerInstanceAPI.from_model(trigger_instance))
        attrs['trigger'] = vars(TriggerAPI.from_model(trigger))
        attrs['trigger_type'] = vars(TriggerTypeAPI.from_model(trigger_type))

    parent = _get_parent_execution(liveaction)
    if parent:
        attrs['parent'] = str(parent.id)

    attrs['log'] = [_create_execution_log_entry(liveaction['status'])]

    execution = ActionExecutionDB(**attrs)
    execution = ActionExecution.add_or_update(execution, publish=False)

    # Update the web_url field in execution. Unfortunately, we need
    # the execution id for constructing the URL which we only get
    # after the model is written to disk.
    execution.web_url = _get_web_url_for_execution(str(execution.id))
    execution = ActionExecution.add_or_update(execution, publish=publish)

    if parent:
        if str(execution.id) not in parent.children:
            parent.children.append(str(execution.id))
            ActionExecution.add_or_update(parent)

    return execution

    def test_crud_partial(self):
        # Create the DB record.
        obj = ActionExecutionAPI(**copy.deepcopy(self.fake_history_subtasks[0]))
        ActionExecution.add_or_update(ActionExecutionAPI.to_model(obj))
        model = ActionExecution.get_by_id(obj.id)
        self.assertEqual(str(model.id), obj.id)
        self.assertDictEqual(model.trigger, {})
        self.assertDictEqual(model.trigger_type, {})
        self.assertDictEqual(model.trigger_instance, {})
        self.assertDictEqual(model.rule, {})
        self.assertDictEqual(model.action, self.fake_history_subtasks[0]['action'])
        self.assertDictEqual(model.runner, self.fake_history_subtasks[0]['runner'])
        doc = copy.deepcopy(self.fake_history_subtasks[0]['liveaction'])
        doc['start_timestamp'] = doc['start_timestamp']
        doc['end_timestamp'] = doc['end_timestamp']
        self.assertDictEqual(model.liveaction, doc)
        self.assertEqual(model.parent, self.fake_history_subtasks[0]['parent'])
        self.assertListEqual(model.children, [])

        # Update the DB record.
        children = [str(bson.ObjectId()), str(bson.ObjectId())]
        model.children = children
        ActionExecution.add_or_update(model)
        model = ActionExecution.get_by_id(obj.id)
        self.assertListEqual(model.children, children)

        # Delete the DB record.
        ActionExecution.delete(model)
        self.assertRaises(StackStormDBObjectNotFoundError, ActionExecution.get_by_id, obj.id)

 def test_chained_executions(self):
     liveaction = LiveActionDB(action='core.chain')
     liveaction, _ = action_service.request(liveaction)
     liveaction = LiveAction.get_by_id(str(liveaction.id))
     self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_FAILED)
     execution = self._get_action_execution(liveaction__id=str(liveaction.id),
                                            raise_exception=True)
     action = action_utils.get_action_by_ref('core.chain')
     self.assertDictEqual(execution.action, vars(ActionAPI.from_model(action)))
     runner = RunnerType.get_by_name(action.runner_type['name'])
     self.assertDictEqual(execution.runner, vars(RunnerTypeAPI.from_model(runner)))
     liveaction = LiveAction.get_by_id(str(liveaction.id))
     self.assertEqual(execution.start_timestamp, liveaction.start_timestamp)
     self.assertEqual(execution.end_timestamp, liveaction.end_timestamp)
     self.assertEqual(execution.result, liveaction.result)
     self.assertEqual(execution.status, liveaction.status)
     self.assertEqual(execution.context, liveaction.context)
     self.assertEqual(execution.liveaction['callback'], liveaction.callback)
     self.assertEqual(execution.liveaction['action'], liveaction.action)
     self.assertGreater(len(execution.children), 0)
     for child in execution.children:
         record = ActionExecution.get(id=child, raise_exception=True)
         self.assertEqual(record.parent, str(execution.id))
         self.assertEqual(record.action['name'], 'local')
         self.assertEqual(record.runner['name'], 'run-local')

 def test_execution_creation_chains(self):
     childliveaction = self.MODELS['liveactions']['childliveaction.yaml']
     child_exec = executions_util.create_execution_object(childliveaction)
     parent_execution_id = childliveaction.context['parent']['execution_id']
     parent_execution = ActionExecution.get_by_id(parent_execution_id)
     child_execs = parent_execution.children
     self.assertTrue(str(child_exec.id) in child_execs)

 def assign_parent(child):
     candidates = [v for k, v in cls.refs.items() if v.action['name'] == 'chain']
     if candidates:
         parent = random.choice(candidates)
         child['parent'] = str(parent.id)
         parent.children.append(child['id'])
         cls.refs[str(parent.id)] = ActionExecution.add_or_update(parent)

def request_cancellation(liveaction, requester):
    """
    Request cancellation of an action execution.

    :return: (liveaction, execution)
    :rtype: tuple
    """
    if liveaction.status == action_constants.LIVEACTION_STATUS_CANCELING:
        return liveaction

    if liveaction.status not in action_constants.LIVEACTION_CANCELABLE_STATES:
        raise Exception(
            'Unable to cancel liveaction "%s" because it is already in a '
            'completed state.' % liveaction.id
        )

    result = {
        'message': 'Action canceled by user.',
        'user': requester
    }

    # Run cancelation sequence for liveaction that is in running state or
    # if the liveaction is operating under a workflow.
    if ('parent' in liveaction.context or
            liveaction.status in action_constants.LIVEACTION_STATUS_RUNNING):
        status = action_constants.LIVEACTION_STATUS_CANCELING
    else:
        status = action_constants.LIVEACTION_STATUS_CANCELED

    liveaction = update_status(liveaction, status, result=result)

    execution = ActionExecution.get(liveaction__id=str(liveaction.id))

    return (liveaction, execution)

 def test_chained_executions(self):
     with mock.patch("st2common.runners.register_runner", mock.MagicMock(return_value=action_chain_runner)):
         liveaction = LiveActionDB(action="executions.chain")
         liveaction, _ = action_service.request(liveaction)
         liveaction = LiveAction.get_by_id(str(liveaction.id))
         self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_FAILED)
         execution = self._get_action_execution(liveaction__id=str(liveaction.id), raise_exception=True)
         action = action_utils.get_action_by_ref("executions.chain")
         self.assertDictEqual(execution.action, vars(ActionAPI.from_model(action)))
         runner = RunnerType.get_by_name(action.runner_type["name"])
         self.assertDictEqual(execution.runner, vars(RunnerTypeAPI.from_model(runner)))
         liveaction = LiveAction.get_by_id(str(liveaction.id))
         self.assertEqual(execution.start_timestamp, liveaction.start_timestamp)
         self.assertEqual(execution.end_timestamp, liveaction.end_timestamp)
         self.assertEqual(execution.result, liveaction.result)
         self.assertEqual(execution.status, liveaction.status)
         self.assertEqual(execution.context, liveaction.context)
         self.assertEqual(execution.liveaction["callback"], liveaction.callback)
         self.assertEqual(execution.liveaction["action"], liveaction.action)
         self.assertGreater(len(execution.children), 0)
         for child in execution.children:
             record = ActionExecution.get(id=child, raise_exception=True)
             self.assertEqual(record.parent, str(execution.id))
             self.assertEqual(record.action["name"], "local")
             self.assertEqual(record.runner["name"], "run-local")