class Action(object):  # pylint: disable=too-many-instance-attributes,too-many-public-methods

    def __init__(self):
        """
        Actions get added to pipelines by calling the
        Pipeline.add_action function. Other Action
        data comes from the parameters. Actions with
        internal pipelines push parameters to actions
        within those pipelines. Parameters are to be
        treated as inmutable.
        """
        # The level of this action within the pipeline. Levels start at one and
        # each pipeline within an command uses a level within the level of the
        # parent pipeline.
        # First command in Outer pipeline: 1
        # First command in pipeline within outer pipeline: 1.1
        # Level is set during pipeline creation and must not be changed
        # subsequently except by RetryCommand.
        self.level = None
        self.pipeline = None
        self.internal_pipeline = None
        self.__parameters__ = {}
        self.__errors__ = []
        self.job = None
        self.logger = logging.getLogger('dispatcher')
        self.__results__ = OrderedDict()
        self.timeout = Timeout(self.name, exception=self.timeout_exception)
        self.max_retries = 1  # unless the strategy or the job parameters change this, do not retry
        self.diagnostics = []
        self.protocols = []  # list of protocol objects supported by this action, full list in job.protocols
        self.connection_timeout = Timeout(self.name, exception=self.timeout_exception)
        self.character_delay = 0
        self.force_prompt = False

    # Section
    section = None
    # public actions (i.e. those who can be referenced from a job file) must
    # declare a 'class-type' name so they can be looked up.
    # summary and description are used to identify instances.
    name = None
    # Used in the pipeline to explain what the commands will attempt to do.
    description = None
    # A short summary of this instance of a class inheriting from Action.  May
    # be None.
    summary = None
    # Exception to raise when this action is timing out
    timeout_exception = JobError

    @property
    def data(self):
        """
        Shortcut to the job.context
        """
        if not self.job:
            return None
        return self.job.context

    @data.setter
    def data(self, value):
        """
        Accepts a dict to be updated in the job.context
        """
        self.job.context.update(value)

    @classmethod
    def select(cls, name):
        for subclass in cls.__subclasses__():  # pylint: disable=no-member
            if subclass.name == name:
                return subclass
        raise JobError("Cannot find action named \"%s\"" % name)

    @property
    def errors(self):
        if self.internal_pipeline:
            return self.__errors__ + self.internal_pipeline.errors
        else:
            return self.__errors__

    @errors.setter
    def errors(self, error):
        if error:
            self.__errors__.append(error)

    @property
    def valid(self):
        return len([x for x in self.errors if x]) == 0

    @property
    def parameters(self):
        """
        All data which this action needs to have available for
        the prepare, run or post_process functions needs to be
        set as a parameter. The parameters will be validated
        during pipeline creation.
        This allows all pipelines to be fully described, including
        the parameters supplied to each action, as well as supporting
        tests on each parameter (like 404 or bad formatting) during
        validation of each action within a pipeline.
        Parameters are static, internal data within each action
        copied directly from the YAML. Dynamic data is held in
        the context available via the parent Pipeline()
        """
        return self.__parameters__

    def __set_parameters__(self, data):
        try:
            self.__parameters__.update(data)
        except ValueError:
            raise LAVABug("Action parameters need to be a dictionary")

        # Overide the duration if needed
        if 'timeout' in self.parameters:
            # preserve existing overrides
            if self.timeout.duration == Timeout.default_duration():
                self.timeout.duration = Timeout.parse(self.parameters['timeout'])
        if 'connection_timeout' in self.parameters:
            self.connection_timeout.duration = Timeout.parse(self.parameters['connection_timeout'])

        # only unit tests should have actions without a pointer to the job.
        if 'failure_retry' in self.parameters and 'repeat' in self.parameters:
            raise JobError("Unable to use repeat and failure_retry, use a repeat block")
        if 'failure_retry' in self.parameters:
            self.max_retries = self.parameters['failure_retry']
        if 'repeat' in self.parameters:
            self.max_retries = self.parameters['repeat']
        if self.job:
            if self.job.device:
                if 'character_delays' in self.job.device:
                    self.character_delay = self.job.device['character_delays'].get(self.section, 0)

    @parameters.setter
    def parameters(self, data):
        self.__set_parameters__(data)
        if self.pipeline:
            for action in self.pipeline.actions:
                action.parameters = self.parameters

    @property
    def results(self):
        """
        Updated dictionary of results for this action.
        """
        return self.__results__

    @results.setter
    def results(self, data):
        try:
            self.__results__.update(data)
        except ValueError:
            raise LAVABug("Action results need to be a dictionary")

    def validate(self):
        """
        This method needs to validate the parameters to the action. For each
        validation that is found, an item should be added to self.errors.
        Validation includes parsing the parameters for this action for
        values not set or values which conflict.
        """
        # Basic checks
        if not self.name:
            self.errors = "%s action has no name set" % self
        # have already checked that self.name is not None, but pylint gets confused.
        if ' ' in self.name:  # pylint: disable=unsupported-membership-test
            self.errors = "Whitespace must not be used in action names, only descriptions or summaries: %s" % self.name

        if '_' in self.name:  # pylint: disable=unsupported-membership-test
            self.errors = "Use - instead of _ in action names: %s" % self.name

        if not self.summary:
            self.errors = "action %s (%s) lacks a summary" % (self.name, self)

        if not self.description:
            self.errors = "action %s (%s) lacks a description" % (self.name, self)

        if not self.section:
            self.errors = "action %s (%s) has no section set" % (self.name, self)

        # Collect errors from internal pipeline actions
        if self.internal_pipeline:
            self.internal_pipeline.validate_actions()

    def populate(self, parameters):
        """
        This method allows an action to add an internal pipeline.
        The parameters are used to configure the internal pipeline on the
        fly.
        """
        pass

    def run_command(self, command_list, allow_silent=False, allow_fail=False, cwd=None):  # pylint: disable=too-many-branches
        """
        Single location for all external command operations on the
        dispatcher, without using a shell and with full structured logging.
        Ensure that output for the YAML logger is a serialisable object
        and strip embedded newlines / whitespace where practical.
        Returns the output of the command (after logging the output)
        Includes default support for proxy settings in the environment.
        Blocks until the command returns then processes & logs the output.

        Caution: take care with the return value as this is highly dependent
        on the command_list and the expected results.

        :param: command_list - the command to run, with arguments
        :param: allow_silent - if True, the command may exit zero with no output
        without being considered to have failed.
        :return: On success (command exited zero), returns the command output.
        If allow_silent is True and the command produced no output, returns True.
        On failure (command exited non-zero), sets self.errors.
        If allow_silent is True, returns False, else returns the command output.
        """
        # FIXME: add option to only check stdout or stderr for failure output
        if not isinstance(command_list, list):
            raise LAVABug("commands to run_command need to be a list")
        log = None
        # nice is assumed to always exist (coreutils)
        command_list = ['nice'] + [str(s) for s in command_list]
        self.logger.debug("%s", ' '.join(command_list))
        try:
            log = subprocess.check_output(command_list, stderr=subprocess.STDOUT,
                                          cwd=cwd)
            log = log.decode('utf-8', errors="replace")  # pylint: disable=redefined-variable-type
        except subprocess.CalledProcessError as exc:
            # the errors property doesn't support removing errors
            errors = []
            if exc.output:
                errors.append(exc.output.strip().decode('utf-8', errors="replace"))
            else:
                errors.append(str(exc))
            msg = '[%s] command %s\nmessage %s\noutput %s\n' % (
                self.name, [i.strip() for i in exc.cmd], str(exc), str(exc).split('\n'))

            # the exception is raised due to a non-zero exc.returncode
            if allow_fail:
                self.logger.info(msg)
                log = exc.output.strip().decode('utf-8', errors="replace")
            else:
                for error in errors:
                    self.errors = error
                self.logger.error(msg)
                # if not allow_fail, fail the command
                return False

        # allow for commands which return no output
        if not log and allow_silent:
            return self.errors == []
        else:
            for line in log.split("\n"):
                self.logger.debug('output: %s', line)
            return log

    def call_protocols(self):
        """
        Actions which support using protocol calls from the job submission use this routine to execute those calls.
        It is up to the action to determine when the protocols are called within the run step of that action.
        The order in which calls are made for any one action is not guaranteed.
        The reply is set in the context data.
        Although actions may have multiple protocol calls in individual tests, use of multiple calls in Strategies
        needs to be avoided to ensure that the individual calls can be easily reused and identified.
        """
        if 'protocols' not in self.parameters:
            return
        for protocol in self.job.protocols:
            if protocol.name not in self.parameters['protocols']:
                # nothing to do for this action with this protocol
                continue
            params = self.parameters['protocols'][protocol.name]
            for call_dict in [call for call in params if 'action' in call and call['action'] == self.name]:
                del call_dict['yaml_line']
                if 'message' in call_dict:
                    del call_dict['message']['yaml_line']
                if 'timeout' in call_dict:
                    del call_dict['timeout']['yaml_line']
                protocol.check_timeout(self.connection_timeout.duration, call_dict)
                self.logger.info("Making protocol call for %s using %s", self.name, protocol.name)
                reply = protocol(call_dict, action=self)
                message = protocol.collate(reply, call_dict)
                if message:
                    self.logger.info("Setting namespace data key %s to %s", message[0], message[1])
                    self.set_namespace_data(
                        action=protocol.name, label=protocol.name, key=message[0], value=message[1])

    def run(self, connection, max_end_time, args=None):
        """
        This method is responsible for performing the operations that an action
        is supposed to do.

        This method usually returns nothing. If it returns anything, that MUST
        be an instance of Connection. That connection will be the one passed on
        to the next action in the pipeline.

        In this classs this method does nothing. It must be implemented by
        subclasses

        :param connection: The Connection object to use to run the steps
        :param max_end_time: The maximum time before this action will timeout.
        :param args: Command and arguments to run
        :raise: Classes inheriting from BaseAction must handle
        all exceptions possible from the command and re-raise
        """
        self.call_protocols()
        if self.internal_pipeline:
            return self.internal_pipeline.run_actions(connection, max_end_time, args)
        if connection:
            connection.timeout = self.connection_timeout
        return connection

    def cleanup(self, connection):
        """
        cleanup will *only* be called after run() if run() raises an exception.
        Use cleanup with any resources that may be left open by an interrupt or failed operation
        such as, but not limited to:

            - open file descriptors
            - mount points
            - error codes

        Use contextmanagers or signal handlers to clean up any resources when there are no errors,
        instead of using cleanup().
        """
        if self.internal_pipeline:
            self.internal_pipeline.cleanup(connection)

    def explode(self):
        """
        serialisation support
        Omit our objects marked as internal by inheriting form InternalObject instead of object,
        e.g. SignalMatch
        """
        data = {}
        attrs = set([attr for attr in dir(self)
                     if not attr.startswith('_') and getattr(self, attr) and not
                     isinstance(getattr(self, attr), types.MethodType) and not
                     isinstance(getattr(self, attr), InternalObject)])

        # noinspection PySetFunctionToLiteral
        for attr in attrs - set([
                'internal_pipeline', 'job', 'logger', 'pipeline',
                'default_fixupdict', 'pattern',
                'parameters', 'SignalDirector', 'signal_director']):
            if attr == 'timeout':
                data['timeout'] = {'duration': self.timeout.duration, 'name': self.timeout.name}
            elif attr == 'connection_timeout':
                data['timeout'] = {'duration': self.timeout.duration, 'name': self.timeout.name}
            elif attr == 'url':
                data['url'] = self.url.geturl()  # pylint: disable=no-member
            elif attr == 'vcs':
                data[attr] = getattr(self, attr).url
            elif attr == 'protocols':
                data['protocols'] = {}
                for protocol in getattr(self, attr):
                    data['protocols'][protocol.name] = {}
                    protocol_attrs = set([attr for attr in dir(protocol)
                                          if not attr.startswith('_') and getattr(protocol, attr) and not
                                          isinstance(getattr(protocol, attr), types.MethodType) and not
                                          isinstance(getattr(protocol, attr), InternalObject)])
                    for protocol_attr in protocol_attrs:
                        if protocol_attr not in ['logger']:
                            data['protocols'][protocol.name][protocol_attr] = getattr(protocol, protocol_attr)
            elif isinstance(getattr(self, attr), OrderedDict):
                data[attr] = dict(getattr(self, attr))
            else:
                data[attr] = getattr(self, attr)
        if 'deployment_data' in self.parameters:
            data['parameters'] = dict()
            data['parameters']['deployment_data'] = self.parameters['deployment_data'].__data__
        return data

    def get_namespace_data(self, action, label, key, deepcopy=True, parameters=None):  # pylint: disable=too-many-arguments
        """
        Get a namespaced data value from dynamic job data using the specified key.
        By default, returns a deep copy of the value instead of a reference to allow actions to
        manipulate lists and dicts based on common data without altering the values used by other actions.
        :param action: Name of the action which set the data or a commonly shared string used to
            correlate disparate actions
        :param label: Arbitrary label used by many actions to sub-divide similar keys with distinct
            values. Can be set to the same as the action.
        :param key: The lookup key for the requested value within the data defined by
            data[namespace][action][label]
        :param deepcopy: If deepcopy is False, the reference is used - meaning that certain operations on the
            namespaced data values other than simple strings will be able to modify the data without calls to
            set_namespace_data.
        :param parameters: Pass parameters when calling get_namespace_data from populate() as the parameters
            will not have been set in the action at that point.
        """
        params = parameters if parameters else self.parameters
        namespace = params['namespace']
        value = self.data.get(namespace, {}).get(action, {}).get(label, {}).get(key, None)  # pylint: disable=no-member
        if value is None:
            return None
        return copy.deepcopy(value) if deepcopy else value

    def set_namespace_data(self, action, label, key, value, parameters=None):  # pylint: disable=too-many-arguments
        """
        Storage for filenames (on dispatcher or on device) and other common data (like labels and ID strings)
        which are set in one Action and used in one or more other Actions elsewhere in the same pipeline.
        :param action: Name of the action which set the data or a commonly shared string used to
            correlate disparate actions
        :param label: Arbitrary label used by many actions to sub-divide similar keys with distinct
            values. Can be set to the same as the action.
        :param key: The lookup key for the requested value within the data defined by
            data[namespace][action][label]
        :param value: The value to set into the namespace data, can be an object.
        :param parameters: Pass parameters when calling get_namespace_data from populate() as the parameters
            will not have been set in the action at that point.
        """
        params = parameters if parameters else self.parameters
        namespace = params['namespace']
        if not label or not key:
            raise LAVABug("Invalid call to set_namespace_data: %s" % action)
        self.data.setdefault(namespace, {})  # pylint: disable=no-member
        self.data[namespace].setdefault(action, {})
        self.data[namespace][action].setdefault(label, {})
        self.data[namespace][action][label][key] = value

    def wait(self, connection, max_end_time=None):
        if not connection:
            return
        if not connection.connected:
            self.logger.debug("Already disconnected")
            return
        if not max_end_time:
            max_end_time = self.timeout.duration + self.timeout.start
        remaining = max_end_time - time.time()
        # FIXME: connection.prompt_str needs to be always a list
        # bootloader_prompt is one which does not get set that way
        # also need functionality to clear the list at times.
        self.logger.debug("%s: Wait for prompt %s (timeout %s)",
                          self.name, connection.prompt_str, seconds_to_str(remaining))
        if self.force_prompt:
            return connection.force_prompt_wait(remaining)
        else:
            return connection.wait(max_end_time)

    def mkdtemp(self, override=None):
        return self.job.mkdtemp(self.name, override=override)

    def _override_action_timeout(self, timeout):
        """
        Only to be called by the Pipeline object, add_action().
        """
        if timeout is None:
            return
        if not isinstance(timeout, dict):
            raise JobError("Invalid timeout %s" % str(timeout))
        self.timeout.duration = Timeout.parse(timeout)

    def _override_connection_timeout(self, timeout):
        """
        Only to be called by the Pipeline object, add_action().
        """
        if timeout is None:
            return
        if not isinstance(timeout, dict):
            raise JobError("Invalid connection timeout %s" % str(timeout))
        self.connection_timeout.duration = Timeout.parse(timeout)

    def log_action_results(self, fail=False):
        if self.results and isinstance(self.logger, YAMLLogger):
            res = "pass"
            if self.errors:
                res = "fail"
            if fail:
                # allows retries without setting errors, which make the job incomplete.
                res = "fail"
            self.logger.results({  # pylint: disable=no-member
                "definition": "lava",
                "namespace": self.parameters.get('namespace', 'common'),
                "case": self.name,
                "level": self.level,
                "duration": "%.02f" % self.timeout.elapsed_time,
                "result": res,
                "extra": self.results})
            self.results.update(
                {
                    'duration': self.timeout.elapsed_time,
                    'timeout': self.timeout.duration,
                    'connection-timeout': self.connection_timeout.duration
                }
            )

    @nottest
    def test_needs_deployment(self, parameters):  # pylint: disable=no-self-use
        if parameters['namespace'] in parameters['test_info']:
            testclasses = parameters['test_info'][parameters['namespace']]
            for testclass in testclasses:
                if testclass['class'].needs_deployment_data():
                    return True
        return False

    @nottest
    def test_has_shell(self, parameters):  # pylint: disable=no-self-use
        if parameters['namespace'] in parameters['test_info']:
            testclasses = parameters['test_info'][parameters['namespace']]
            for testclass in testclasses:
                if testclass['class'].has_shell():
                    return True
        return False

    @nottest
    def test_needs_overlay(self, parameters):  # pylint: disable=no-self-use
        if parameters['namespace'] in parameters['test_info']:
            testclasses = parameters['test_info'][parameters['namespace']]
            for testclass in testclasses:
                if testclass['class'].needs_overlay():
                    return True
        return False class InfrastructureError(LAVAError):
    """
    Exceptions based on an error raised by a component of the
    test which is neither the LAVA dispatcher code nor the
    code being executed on the device under test. This includes
    errors arising from the device (like the arndale SD controller
    issue) and errors arising from the hardware to which the device
    is connected (serial console connection, ethernet switches or
    internet connection beyond the control of the device under test).

    Use LAVABug for errors arising from bugs in LAVA code.
    """
    error_help = "InfrastructureError: The Infrastructure is not working " \
                 "correctly. Please report this error to LAVA admins."
    error_type = "Infrastructure" class JobError(LAVAError):
    """
    An Error arising from the information supplied as part of the TestJob
    e.g. HTTP404 on a file to be downloaded as part of the preparation of
    the TestJob or a download which results in a file which tar or gzip
    does not recognise.
    """
    error_help = "JobError: Your job cannot terminate cleanly."
    error_type = "Job" class LAVABug(LAVAError):
    """
    An error that is raised when an un-expected error is catched. Only happen
    when a bug is encountered.
    """
    error_help = "LAVABug: This is probably a bug in LAVA, please report it."
    error_type = "Bug" class Pipeline(object):  # pylint: disable=too-many-instance-attributes
    """
    Pipelines ensure that actions are run in the correct sequence whilst
    allowing for retries and other requirements.
    When an action is added to a pipeline, the level of that action within
    the overall job is set along with the formatter and output filename
    of the per-action log handler.
    """
    def __init__(self, parent=None, job=None, parameters=None):
        self.actions = []
        self.parent = None
        self.parameters = {} if parameters is None else parameters
        self.job = job
        if parent is not None:
            # parent must be an Action
            if not isinstance(parent, Action):
                raise LAVABug("Internal pipelines need an Action as a parent")
            if not parent.level:
                raise LAVABug("Tried to create a pipeline using a parent action with no level set.")
            self.parent = parent

    def _check_action(self, action):  # pylint: disable=no-self-use
        if not action or not issubclass(type(action), Action):
            raise LAVABug("Only actions can be added to a pipeline: %s" % action)
        # if isinstance(action, DiagnosticAction):
        #     raise LAVABug("Diagnostic actions need to be triggered, not added to a pipeline.")
        if not action:
            raise LAVABug("Unable to add empty action to pipeline")

    def add_action(self, action, parameters=None):  # pylint: disable=too-many-branches
        self._check_action(action)
        self.actions.append(action)
        # FIXME: if this is only happening in unit test, this has to be fixed later on
        # should only be None inside the unit tests
        action.job = self.job
        if self.parent:  # action
            self.parent.pipeline = self
            action.level = "%s.%s" % (self.parent.level, len(self.actions))
            action.section = self.parent.section
        else:
            action.level = "%s" % (len(self.actions))

        # Use the pipeline parameters if the function was walled without
        # parameters.
        if parameters is None:
            parameters = self.parameters
        # if the action has an internal pipeline, initialise that here.
        action.populate(parameters)

        # Compute the timeout
        timeouts = []
        # FIXME: Only needed for the auto-tests
        if self.job is not None:
            if self.job.device is not None:
                # First, the device level overrides
                timeouts.append(self.job.device.get('timeouts', {}))
            # Then job level overrides
            timeouts.append(self.job.parameters.get('timeouts', {}))

        # pylint: disable=invalid-name

        def dict_merge_get(dicts, key):
            value = None
            for d in dicts:
                value = d.get(key, value)
            return value

        def subdict_merge_get(dicts, key, subkey):
            value = None
            for d in dicts:
                value = d.get(key, {}).get(subkey, value)
            return value

        # Set the timeout. The order is:
        # 1/ the global action timeout
        # 2/ the individual action timeout
        # 3/ the action block timeout
        # pylint: disable=protected-access
        action._override_action_timeout(dict_merge_get(timeouts, 'action'))
        action._override_action_timeout(subdict_merge_get(timeouts, 'actions', action.name))
        action._override_action_timeout(parameters.get('timeout', None))

        action._override_connection_timeout(dict_merge_get(timeouts, 'connection'))
        action._override_connection_timeout(subdict_merge_get(timeouts, 'connections', action.name))

        action.parameters = parameters

    def describe(self, verbose=True):
        """
        Describe the current pipeline, recursing through any
        internal pipelines.
        :return: a recursive dictionary
        """
        desc = []
        for action in self.actions:
            if verbose:
                current = action.explode()
            else:
                cls = str(type(action))[8:-2].replace('lava_dispatcher.', '')
                current = {'class': cls, 'name': action.name}
            if action.pipeline is not None:
                current['pipeline'] = action.pipeline.describe(verbose)
            desc.append(current)
        return desc

    @property
    def errors(self):
        sub_action_errors = [a.errors for a in self.actions]
        if not sub_action_errors:  # allow for jobs with no actions
            return []
        return reduce(lambda a, b: a + b, sub_action_errors)

    def validate_actions(self):
        for action in self.actions:
            action.validate()

        # If this is the root pipeline, raise the errors
        if self.parent is None and self.errors:
            raise JobError("Invalid job data: %s\n" % self.errors)

    def cleanup(self, connection):
        """
        Recurse through internal pipelines running action.cleanup(),
        in order of the pipeline levels.
        """
        for child in self.actions:
            try:
                child.cleanup(connection)
            except Exception as exc:
                # Just log the exception and continue the cleanup
                child.logger.error("Failed to clean after action '%s': %s", child.name, str(exc))
                child.logger.exception(traceback.format_exc())

    def _diagnose(self, connection):
        """
        Pipeline Jobs have a number of Diagnostic classes registered - all
        supported DiagnosticAction classes should be registered with the Job.
        If an Action.run() function reports a JobError or InfrastructureError,
        the Pipeline calls Job.diagnose(). The job iterates through the DiagnosticAction
        classes declared in the diagnostics list, checking if the trigger classmethod
        matches the requested complaint. Matching diagnostics are run using the current Connection.
        Actions generate a complaint by appending the return of the trigger classmethod
        to the triggers list of the Job. This can be done at any point prior to the
        exception being raised in the run function.
        The trigger list is cleared after each diagnostic operation is complete.
        """
        for complaint in self.job.triggers:
            diagnose = self.job.diagnose(complaint)
            if diagnose:
                connection = diagnose.run(connection, None)
            else:
                raise LAVABug("No diagnosis for trigger %s" % complaint)
        self.job.triggers = []
        # Diagnosis is not allowed to alter the connection, do not use the return value.
        return None

    def run_actions(self, connection, max_end_time, args=None):  # pylint: disable=too-many-branches,too-many-statements,too-many-locals
        for action in self.actions:
            failed = False
            namespace = action.parameters.get('namespace', 'common')
            # Begin the action
            try:
                with action.timeout(max_end_time) as action_max_end_time:
                    # Add action start timestamp to the log message
                    # Log in INFO for root actions and in DEBUG for the other actions
                    timeout = seconds_to_str(action_max_end_time - action.timeout.start)
                    msg = 'start: %s %s (timeout %s) [%s]' % (
                        action.level, action.name, timeout, namespace)
                    if self.parent is None:
                        action.logger.info(msg)
                    else:
                        action.logger.debug(msg)

                    new_connection = action.run(connection,
                                                action_max_end_time, args)
            except LAVAError as exc:
                action.logger.exception(str(exc))
                # allows retries without setting errors, which make the job incomplete.
                failed = True
                action.results = {'fail': str(exc)}
                self._diagnose(connection)
                raise
            except Exception as exc:
                action.logger.exception(traceback.format_exc())
                # allows retries without setting errors, which make the job incomplete.
                failed = True
                action.results = {'fail': str(exc)}
                # Raise a LAVABug that will be correctly classified later
                raise LAVABug(str(exc))
            finally:
                # Add action end timestamp to the log message
                duration = round(action.timeout.elapsed_time)
                msg = "end: %s %s (duration %s) [%s]" % (
                    action.level, action.name,
                    seconds_to_str(duration), namespace)
                if self.parent is None:
                    action.logger.info(msg)
                else:
                    action.logger.debug(msg)
                # set results including retries and failed actions
                action.log_action_results(fail=failed)

            if new_connection:
                connection = new_connection
        return connection class OverlayAction(DeployAction):
    """
    Creates a temporary location into which the lava test shell scripts are installed.
    The location remains available for the testdef actions to populate
    Multinode and LMP actions also populate the one location.
    CreateOverlay then creates a tarball of that location in the output directory
    of the job and removes the temporary location.
    ApplyOverlay extracts that tarball onto the image.

    Deployments which are for a job containing a 'test' action will have
    a TestDefinitionAction added to the job pipeline by this Action.

    The resulting overlay needs to be applied separately and custom classes
    exist for particular deployments, so that the overlay can be applied
    whilst the image is still mounted etc.

    This class handles parts of the overlay which are independent
    of the content of the test definitions themselves. Other
    overlays are handled by TestDefinitionAction.
    """

    name = "lava-overlay"
    description = "add lava scripts during deployment for test shell use"
    summary = "overlay the lava support scripts"

    def __init__(self):
        super(OverlayAction, self).__init__()
        self.lava_test_dir = os.path.realpath(
            '%s/../../lava_test_shell' % os.path.dirname(__file__))
        self.scripts_to_copy = []
        # 755 file permissions
        self.xmod = stat.S_IRWXU | stat.S_IXGRP | stat.S_IRGRP | stat.S_IXOTH | stat.S_IROTH
        self.target_mac = ''
        self.target_ip = ''
        self.probe_ip = ''
        self.probe_channel = ''

    def validate(self):
        super(OverlayAction, self).validate()
        self.scripts_to_copy = sorted(glob.glob(os.path.join(self.lava_test_dir, 'lava-*')))
        # Distro-specific scripts override the generic ones
        if not self.test_needs_overlay(self.parameters):
            return
        lava_test_results_dir = self.parameters['deployment_data']['lava_test_results_dir']
        lava_test_results_dir = lava_test_results_dir % self.job.job_id
        self.set_namespace_data(action='test', label='results', key='lava_test_results_dir',
                                value=lava_test_results_dir)
        lava_test_sh_cmd = self.parameters['deployment_data']['lava_test_sh_cmd']
        self.set_namespace_data(action='test', label='shared', key='lava_test_sh_cmd',
                                value=lava_test_sh_cmd)

        # Add distro support scripts
        distro = self.parameters['deployment_data']['distro']
        distro_support_dir = '%s/distro/%s' % (self.lava_test_dir, distro)
        self.scripts_to_copy += sorted(glob.glob(os.path.join(distro_support_dir,
                                                              'lava-*')))

        if not self.scripts_to_copy:
            self.errors = "Unable to locate lava_test_shell support scripts."
        if 'parameters' in self.job.device:
            if 'interfaces' in self.job.device['parameters']:
                if 'target' in self.job.device['parameters']['interfaces']:
                    self.target_mac = self.job.device['parameters']['interfaces']['target'].get('mac', '')
                    self.target_ip = self.job.device['parameters']['interfaces']['target'].get('ip', '')
        for device in self.job.device.get('static_info', []):
            if 'probe_channel' in device and 'probe_ip' in device:
                self.probe_channel = device['probe_channel']
                self.probe_ip = device['probe_ip']
                break

    def populate(self, parameters):
        self.internal_pipeline = Pipeline(parent=self, job=self.job, parameters=parameters)
        if self.test_needs_overlay(parameters):
            if any('ssh' in data for data in self.job.device['actions']['deploy']['methods']):
                # only devices supporting ssh deployments add this action.
                self.internal_pipeline.add_action(SshAuthorize())
            self.internal_pipeline.add_action(VlandOverlayAction())
            self.internal_pipeline.add_action(MultinodeOverlayAction())
            self.internal_pipeline.add_action(TestDefinitionAction())
            self.internal_pipeline.add_action(CompressOverlay())
            self.internal_pipeline.add_action(PersistentNFSOverlay())  # idempotent

    def run(self, connection, max_end_time, args=None):  # pylint: disable=too-many-locals
        """
        Check if a lava-test-shell has been requested, implement the overlay
        * create test runner directories beneath the temporary location
        * copy runners into test runner directories
        """
        tmp_dir = self.mkdtemp()
        namespace = self.parameters.get('namespace', None)
        if namespace:
            if namespace not in get_test_action_namespaces(self.job.parameters):
                self.logger.info("[%s] skipped %s - no test action.", namespace, self.name)
                return connection
        self.set_namespace_data(action='test', label='shared', key='location', value=tmp_dir)
        lava_test_results_dir = self.get_namespace_data(action='test', label='results', key='lava_test_results_dir')
        if not lava_test_results_dir:
            raise LAVABug("Unable to identify top level lava test directory")
        shell = self.get_namespace_data(action='test', label='shared', key='lava_test_sh_cmd')
        self.logger.debug("[%s] Preparing overlay tarball in %s", namespace, tmp_dir)
        lava_path = os.path.abspath("%s/%s" % (tmp_dir, lava_test_results_dir))
        for runner_dir in ['bin', 'tests', 'results']:
            # avoid os.path.join as lava_test_results_dir startswith / so location is *dropped* by join.
            path = os.path.abspath("%s/%s" % (lava_path, runner_dir))
            if not os.path.exists(path):
                os.makedirs(path, 0o755)
                self.logger.debug("makedir: %s", path)
        for fname in self.scripts_to_copy:
            with open(fname, 'r') as fin:
                foutname = os.path.basename(fname)
                output_file = '%s/bin/%s' % (lava_path, foutname)
                if "distro" in fname:
                    distribution = os.path.basename(os.path.dirname(fname))
                    self.logger.debug("Updating %s (%s)", output_file, distribution)
                else:
                    self.logger.debug("Creating %s", output_file)
                with open(output_file, 'w') as fout:
                    fout.write("#!%s\n\n" % shell)
                    if foutname == 'lava-target-mac':
                        fout.write("TARGET_DEVICE_MAC='%s'\n" % self.target_mac)
                    if foutname == 'lava-target-ip':
                        fout.write("TARGET_DEVICE_IP='%s'\n" % self.target_ip)
                    if foutname == 'lava-probe-ip':
                        fout.write("PROBE_DEVICE_IP='%s'\n" % self.probe_ip)
                    if foutname == 'lava-probe-channel':
                        fout.write("PROBE_DEVICE_CHANNEL='%s'\n" % self.probe_channel)
                    if foutname == 'lava-target-storage':
                        fout.write('LAVA_STORAGE="\n')
                        for method in self.job.device.get('storage_info', [{}]):
                            for key, value in method.items():
                                if key == 'yaml_line':
                                    continue
                                self.logger.debug("storage methods:\t%s\t%s", key, value)
                                fout.write(r"\t%s\t%s\n" % (key, value))
                        fout.write('"\n')
                    fout.write(fin.read())
                    os.fchmod(fout.fileno(), self.xmod)

        # Generate the file containing the secrets
        if 'secrets' in self.job.parameters:
            self.logger.debug("Creating %s/secrets", lava_path)
            with open(os.path.join(lava_path, 'secrets'), 'w') as fout:
                for key, value in self.job.parameters['secrets'].items():
                    if key == 'yaml_line':
                        continue
                    fout.write("%s=%s\n" % (key, value))

        connection = super(OverlayAction, self).run(connection, max_end_time, args)
        return connection RAMDISK_FNAME = 'ramdisk.cpio' UBOOT_DEFAULT_HEADER_LENGTH = 64 @contextmanager
def chdir(path):
    pwd = None
    try:
        pwd = os.getcwd()
        os.chdir(path)
        yield
    finally:
        if pwd is not None:
            os.chdir(pwd) def add_late_command(preseedfile, extra_command):
    added = False
    with open(preseedfile, "r") as pf:
        lines = pf.readlines()
        pf.close()
    endstring = '\\\n'
    while endsin(lines, endstring):
        for linenum, data in enumerate(lines):
            if endsin(data, endstring):
                lines[linenum] = lines[linenum].replace(endstring, '') + lines[linenum + 1]
                del lines[linenum + 1]
    for linenum, data in enumerate(lines):
        if re.match("d-i preseed/late_command string(.*)", data):
            # late_command already exists, append to it
            append_line = "; " + extra_command + "\n"
            lines[linenum] = lines[linenum].rstrip(' ;\n') + append_line
            added = True
    if not added:
        append_line = extra_command + "\n"
        lines.append("d-i preseed/late_command string " + append_line)

    with open(preseedfile, "w") as pf:
        for line in lines:
            pf.write(line)
        pf.close() def add_to_kickstart(preseedfile, extra_command):
    with open(preseedfile, 'a') as pf:
        pf.write('\n')
        pf.write('%post\n')
        pf.write('exec < /dev/console > /dev/console\n')
        pf.write(extra_command + '\n')
        pf.write('%end\n')
    pf.close() def lxc_path(dispatcher_config):
    """
    Returns LXC_PATH which is a constant, unless a dispatcher specific path is
    configured via lxc_path key in dispatcher_config.
    """
    try:
        return dispatcher_config["lxc_path"]
    except (KeyError, TypeError):
        return LXC_PATH def mkdtemp(autoremove=True, basedir='/tmp'):
    """
    returns a temporary directory that's deleted when the process exits

    """
    tmpdir = tempfile.mkdtemp(dir=basedir)
    os.chmod(tmpdir, 0o755)
    if autoremove:
        atexit.register(rmtree, tmpdir)
    return tmpdir @replace_exception(RuntimeError, JobError)
def prepare_guestfs(output, overlay, size):
    """
    Applies the overlay, offset by one directory.
    This allows the booted device to mount at the
    original lava directory and retain the same path
    as if the overlay was unpacked directly into the
    image.
    :param output: filename of the temporary device
    :param overlay: tarball of the lava test shell overlay.
    :param size: size of the filesystem in Mb
    :return blkid of the guest device
    """
    guest = guestfs.GuestFS(python_return_dict=True)
    guest.disk_create(output, "qcow2", size * 1024 * 1024)
    guest.add_drive_opts(output, format="qcow2", readonly=False)
    guest.launch()
    devices = guest.list_devices()
    if len(devices) != 1:
        raise InfrastructureError("Unable to prepare guestfs")
    guest_device = devices[0]
    guest.mke2fs(guest_device, label='LAVA')
    # extract to a temp location
    tar_output = mkdtemp()
    # Now mount the filesystem so that we can add files.
    guest.mount(guest_device, "/")
    tarball = tarfile.open(overlay)
    tarball.extractall(tar_output)
    guest_dir = mkdtemp()
    guest_tar = os.path.join(guest_dir, 'guest.tar')
    root_tar = tarfile.open(guest_tar, 'w')
    for topdir in os.listdir(tar_output):
        for dirname in os.listdir(os.path.join(tar_output, topdir)):
            root_tar.add(os.path.join(tar_output, topdir, dirname), arcname=dirname)
    root_tar.close()
    guest.tar_in(guest_tar, '/')
    os.unlink(guest_tar)
    guest.umount(guest_device)
    return guest.blkid(guest_device)['UUID'] @replace_exception(RuntimeError, JobError)
def copy_in_overlay(image, root_partition, overlay):
    """
    Mounts test image partition as specified by the test
    writer and extracts overlay at the root, if root_partition
    is None the image is handled as a filesystem instead of
    partitioned image.
    """
    guest = guestfs.GuestFS(python_return_dict=True)
    guest.add_drive(image)
    guest.launch()

    if root_partition:
        partitions = guest.list_partitions()
        if not partitions:
            raise InfrastructureError("Unable to prepare guestfs")
        guest_partition = partitions[root_partition]
        guest.mount(guest_partition, '/')
    else:
        devices = guest.list_devices()
        if not devices:
            raise InfrastructureError("Unable to prepare guestfs")
        guest.mount(devices[0], '/')

    # FIXME: max message length issues when using tar_in
    # on tar.gz.  Works fine with tar so decompressing
    # overlay first.
    if os.path.exists(overlay[:-3]):
        os.unlink(overlay[:-3])
    decompressed_overlay = decompress_file(overlay, 'gz')
    guest.tar_in(decompressed_overlay, '/')

    if root_partition:
        guest.umount(guest_partition)
    else:
        guest.umount(devices[0]) @replace_exception(RuntimeError, JobError)
def copy_overlay_to_sparse_fs(image, overlay):
    """copy_overlay_to_sparse_fs
    """
    ext4_img = image + '.ext4'
    logger = logging.getLogger('dispatcher')
    guest = guestfs.GuestFS(python_return_dict=True)

    # Check if the given image is an Android sparse image
    if not is_sparse_image(image):
        raise JobError("Image is not an Android sparse image: %s" % image)

    subprocess.check_output(['/usr/bin/simg2img', image, ext4_img],
                            stderr=subprocess.STDOUT)
    guest.add_drive(ext4_img)
    guest.launch()
    devices = guest.list_devices()
    if not devices:
        raise InfrastructureError("Unable to prepare guestfs")
    guest.mount(devices[0], '/')
    # FIXME: max message length issues when using tar_in
    # on tar.gz.  Works fine with tar so decompressing
    # overlay first.
    if os.path.exists(overlay[:-3]):
        os.unlink(overlay[:-3])
    decompressed_overlay = decompress_file(overlay, 'gz')
    guest.tar_in(decompressed_overlay, '/')
    # Check if we have space left on the mounted image.
    output = guest.df()
    logger.debug(output)
    device, size, used, available, percent, mountpoint = output.split(
        "\n")[1].split()
    guest.umount(devices[0])
    if int(available) is 0 or percent == '100%':
        raise JobError("No space in image after applying overlay: %s" % image)
    subprocess.check_output(['/usr/bin/img2simg', ext4_img, image],
                            stderr=subprocess.STDOUT)
    os.remove(ext4_img) def which(path, match=os.path.isfile):
    exefile = _which_check(path, match)
    if not exefile:
        raise InfrastructureError("Cannot find command '%s' in $PATH" % path)

    if os.stat(exefile).st_mode & S_IXUSR != S_IXUSR:
        raise InfrastructureError("Cannot execute '%s' at '%s'" % (path, exefile))

    return exefile def compress_file(infile, compression):
    if not compression:
        return infile
    if compression not in compress_command_map.keys():
        raise JobError("Cannot find shell command to compress: %s" % compression)

    # Check that the command does exists
    which(compress_command_map[compression].split(" ")[0])

    with chdir(os.path.dirname(infile)):
        cmd = "%s %s" % (compress_command_map[compression], infile)
        try:
            # safe to use shell=True here, no external arguments
            subprocess.check_output(cmd, shell=True)
            return "%s.%s" % (infile, compression)
        except (OSError, subprocess.CalledProcessError) as exc:
            raise InfrastructureError('unable to compress file %s: %s' % (infile, exc)) def decompress_file(infile, compression):
    if not compression:
        return infile
    if compression not in decompress_command_map.keys():
        raise JobError("Cannot find shell command to decompress: %s" % compression)

    # Check that the command does exists
    which(decompress_command_map[compression])

    with chdir(os.path.dirname(infile)):
        cmd = "%s %s" % (decompress_command_map[compression], infile)
        outfile = infile
        if infile.endswith(compression):
            outfile = infile[:-(len(compression) + 1)]
        try:
            # safe to use shell=True here, no external arguments
            subprocess.check_output(cmd, shell=True)
            return outfile
        except (OSError, subprocess.CalledProcessError) as exc:
            raise InfrastructureError('unable to decompress file %s: %s' % (infile, exc)) def untar_file(infile, outdir, member=None, outfile=None):
    try:
        tar = tarfile.open(infile)
        if member:
            file_obj = tar.extractfile(member)
            target = open(outfile, 'wb')
            target.write(file_obj.read())
            target.close()
            file_obj.close()
            tar.close()
        else:
            tar.extractall(outdir)
            tar.close()
    except tarfile.TarError as exc:
        raise JobError("Unable to unpack %s: %s" % (infile, str(exc))) def substitute(command_list, dictionary):
    """
    Replace markup in the command_list which matches a key in the dictionary with the
    value of that key in the dictionary. Empty values leave the item unchanged.
    Markup needs to be safe to use in the final command as there is no guarantee that
    any dictionary will replace all markup in the command_list.
    arguments: command_list - a list of strings
               dictionary - a dictionary of keys which match some of the strings with values
                            to replace for the key in the string.
    """
    parsed = []
    for line in command_list:
        for key, value in dictionary.items():
            if not key or not value:
                continue
            line = line.replace(key, value)
        parsed.append(line)
    return parsed def dispatcher_ip(dispatcher_config):
    """
    Retrieves the IP address of the interface associated
    with the current default gateway.
    """
    try:
        return dispatcher_config["dispatcher_ip"]
    except (KeyError, TypeError):
        pass
    gateways = netifaces.gateways()
    if 'default' not in gateways:
        raise InfrastructureError("Unable to find dispatcher 'default' gateway")
    iface = gateways['default'][netifaces.AF_INET][1]
    addr = netifaces.ifaddresses(iface)
    return addr[netifaces.AF_INET][0]['addr'] class PrepareKernelAction(Action):
    """
    Populate the pipeline with a kernel conversion action, if needed
    """

    name = "prepare-kernel"
    description = "populates the pipeline with a kernel conversion action"
    summary = "add a kernel conversion"

    def populate(self, parameters):
        self.internal_pipeline = Pipeline(parent=self, job=self.job, parameters=parameters)
        # the logic here can be upgraded in future if needed with more parameters to the deploy.
        methods = self.job.device['actions']['boot']['methods']
        if 'u-boot' in methods:
            self.internal_pipeline.add_action(UBootPrepareKernelAction())
        elif 'depthcharge' in methods:
            self.internal_pipeline.add_action(PrepareFITAction())
import os
import shutil
import subprocess
from lava_dispatcher.action import (
    Action,
    InfrastructureError,
    JobError,
    LAVABug,
    Pipeline,
)
from lava_dispatcher.actions.deploy.overlay import OverlayAction
from lava_dispatcher.utils.constants import (
    RAMDISK_FNAME,
    UBOOT_DEFAULT_HEADER_LENGTH,
)
from lava_dispatcher.utils.contextmanager import chdir
from lava_dispatcher.utils.installers import (
    add_late_command,
    add_to_kickstart
)
from lava_dispatcher.utils.filesystem import (
    lxc_path,
    mkdtemp,
    prepare_guestfs,
    copy_in_overlay,
    copy_overlay_to_sparse_fs,
)
from lava_dispatcher.utils.shell import which
from lava_dispatcher.utils.compression import (
    compress_file,
    decompress_file,
    untar_file
)
from lava_dispatcher.utils.strings import substitute
from lava_dispatcher.utils.network import dispatcher_ip
from lava_dispatcher.actions.deploy.prepare import PrepareKernelAction

    def run(self, connection, max_end_time, args=None):
        if not self.parameters.get('ramdisk', None):  # idempotency
            return connection
        ramdisk = self.get_namespace_data(action='download-action', label='ramdisk', key='file')
        if self.skip:
            self.logger.info("Not extracting ramdisk.")
            suffix = self.get_namespace_data(action='tftp-deploy', label='tftp', key='suffix')
            filename = os.path.join(suffix, "ramdisk", os.path.basename(ramdisk))
            # declare the original ramdisk as the name to be used later.
            self.set_namespace_data(action='compress-ramdisk', label='file', key='ramdisk', value=filename)
            return
        connection = super(ExtractRamdisk, self).run(connection, max_end_time, args)
        ramdisk_dir = self.mkdtemp()
        extracted_ramdisk = os.path.join(ramdisk_dir, 'ramdisk')
        os.mkdir(extracted_ramdisk, 0o755)
        compression = self.parameters['ramdisk'].get('compression', None)
        suffix = ""
        if compression:
            suffix = ".%s" % compression
        ramdisk_compressed_data = os.path.join(ramdisk_dir, RAMDISK_FNAME + suffix)
        if self.parameters['ramdisk'].get('header', None) == 'u-boot':
            cmd = ('dd if=%s of=%s ibs=%s skip=1' % (
                ramdisk, ramdisk_compressed_data, UBOOT_DEFAULT_HEADER_LENGTH)).split(' ')
            try:
                self.run_command(cmd)
            except Exception:
                raise LAVABug('Unable to remove uboot header: %s' % ramdisk)
        else:
            # give the file a predictable name
            shutil.move(ramdisk, ramdisk_compressed_data)
        ramdisk_data = decompress_file(ramdisk_compressed_data, compression)

        with chdir(extracted_ramdisk):
            cmd = ('cpio -iud -F %s' % ramdisk_data).split(' ')
            if not self.run_command(cmd):
                raise JobError('Unable to extract cpio archive: %s - missing header definition (i.e. u-boot)?' % ramdisk_data)

        # tell other actions where the unpacked ramdisk can be found
        self.set_namespace_data(action=self.name, label='extracted_ramdisk', key='directory', value=extracted_ramdisk)
        self.set_namespace_data(action=self.name, label='ramdisk_file', key='file', value=ramdisk_data)
        return connection


class CompressRamdisk(Action):
    """
     recreate ramdisk, with overlay in place
    """

    name = "compress-ramdisk"
    description = "recreate a ramdisk with the overlay applied."
    summary = "compress ramdisk with overlay"

    def __init__(self):
        super(CompressRamdisk, self).__init__()
        self.mkimage_arch = None
        self.add_header = None
        self.skip = False

    def validate(self):
        super(CompressRamdisk, self).validate()
        if not self.parameters.get('ramdisk', None):  # idempotency
            return
        if not self.parameters['ramdisk'].get('install_modules', True) and \
                not self.parameters['ramdisk'].get('install_overlay', True):
            self.skip = True
            return
        if 'parameters' in self.job.device['actions']['deploy']:
            self.add_header = self.job.device['actions']['deploy']['parameters'].get('add_header', None)
            if self.add_header is not None:
                if self.add_header == 'u-boot':
                    which('mkimage')
                    if 'mkimage_arch' not in self.job.device['actions']['deploy']['parameters']:
                        self.errors = "Missing architecture for uboot mkimage support (mkimage_arch in deploy parameters)"
                        return
                    self.mkimage_arch = self.job.device['actions']['deploy']['parameters']['mkimage_arch']
                else:
                    self.errors = "ramdisk: add_header: unknown header type"

    def run(self, connection, max_end_time, args=None):  # pylint: disable=too-many-locals
        if not self.parameters.get('ramdisk', None):  # idempotency
            return connection
        if self.skip:
            return connection
        connection = super(CompressRamdisk, self).run(connection, max_end_time, args)
        ramdisk_dir = self.get_namespace_data(
            action='extract-overlay-ramdisk', label='extracted_ramdisk', key='directory')
        ramdisk_data = self.get_namespace_data(
            action='extract-overlay-ramdisk', label='ramdisk_file', key='file')
        if not ramdisk_dir:
            raise LAVABug("Unable to find unpacked ramdisk")
        if not ramdisk_data:
            raise LAVABug("Unable to find ramdisk directory")
        if self.parameters.get('preseed', None):
            if self.parameters["deployment_data"].get("preseed_to_ramdisk", None):
                # download action must have completed to get this far
                # some installers (centos) cannot fetch the preseed file via tftp.
                # Instead, put the preseed file into the ramdisk using a given name
                # from deployment_data which we can use in the boot commands.
                filename = self.parameters["deployment_data"]["preseed_to_ramdisk"]
                self.logger.info("Copying preseed file into ramdisk: %s", filename)
                shutil.copy(self.get_namespace_data(
                    action='download-action', label='preseed',
                    key='file'), os.path.join(ramdisk_dir, filename))
                self.set_namespace_data(action=self.name, label='file', key='preseed_local', value=filename)

        with chdir(ramdisk_dir):
            self.logger.info("Building ramdisk %s containing %s",
                             ramdisk_data, ramdisk_dir)
            cmd = "find . | cpio --create --format='newc' > %s" % ramdisk_data
            try:
                # safe to use shell=True here, no external arguments
                log = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT)
            except OSError as exc:
                raise InfrastructureError('Unable to create cpio filesystem: %s' % exc)
            # lazy-logging would mean that the quoting of cmd causes invalid YAML
            self.logger.debug("%s\n%s" % (cmd, log))  # pylint: disable=logging-not-lazy

            # we need to compress the ramdisk with the same method is was submitted with
            compression = self.parameters['ramdisk'].get('compression', None)

            final_file = compress_file(ramdisk_data, compression)