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 class UBootFactory(Factory):  # pylint: disable=too-few-public-methods
class TestUbootAction(StdoutTestCase):  # pylint: disable=too-many-public-methods
class TestKernelConversion(StdoutTestCase):
class TestOverlayCommands(TestUbootAction):  # pylint: disable=too-many-public-methods
    def create_bbb_job(self, filename):  # pylint: disable=no-self-use
    def create_x15_job(self, filename):  # pylint: disable=no-self-use
    def create_juno_job(self, filename):  # pylint: disable=no-self-use
    def create_zcu102_job(self, filename):  # pylint: disable=no-self-use
    def setUp(self):
    def test_simulated_action(self):
    def test_tftp_pipeline(self):
    def test_device_bbb(self):
    def test_uboot_action(self):
    def test_fastboot_uboot(self):  # pylint: disable=too-many-locals
    def test_x15_uboot_nfs(self):  # pylint: disable=too-many-locals
    def test_juno_uboot_nfs(self):
    def test_overlay_action(self):  # pylint: disable=too-many-locals
    def test_boot_commands(self):
    def test_nbd_boot(self):
    def test_transfer_media(self):
    def test_download_action(self):
    def test_reset_actions(self):
    def test_secondary_media(self):
    def test_xz_nfs(self):
    def test_prefix(self):
    def test_zcu102(self):
    def setUp(self):
    def test_zimage_bootz(self):
    def test_image(self):
    def test_uimage(self):
    def test_zimage_nobootz(self):
    def test_uimage_boot_type(self):
    def test_combined_ramdisk_nfs(self):
    def test_ramdisk_nfs_nomodules(self): class UBootAction(BootAction):
    """
    Wraps the Retry Action to allow for actions which precede
    the reset, e.g. Connect.
    """

    name = "uboot-action"
    description = "interactive uboot action"
    summary = "pass uboot commands"

    def validate(self):
        super(UBootAction, self).validate()
        if 'type' in self.parameters:
            self.logger.warning("Specifying a type in the boot action is deprecated. "
                                "Please specify the kernel type in the deploy parameters.")

    def populate(self, parameters):
        self.internal_pipeline = Pipeline(parent=self, job=self.job, parameters=parameters)
        # customize the device configuration for this job
        self.internal_pipeline.add_action(UBootSecondaryMedia())
        self.internal_pipeline.add_action(BootloaderCommandOverlay())
        self.internal_pipeline.add_action(ConnectDevice())
        self.internal_pipeline.add_action(UBootRetry()) class UBootRetry(BootAction):

    name = "uboot-retry"
    description = "interactive uboot retry action"
    summary = "uboot commands with retry"

    def populate(self, parameters):
        self.internal_pipeline = Pipeline(parent=self, job=self.job, parameters=parameters)
        self.method_params = self.job.device['actions']['boot']['methods']['u-boot']['parameters']
        self.usb_mass_device = self.method_params.get('uboot_mass_storage_device', None)
        # establish a new connection before trying the reset
        self.internal_pipeline.add_action(ResetDevice())
        self.internal_pipeline.add_action(BootloaderInterruptAction())
        if self.method_params.get('uboot_ums_flash', False):
            self.internal_pipeline.add_action(BootloaderCommandsAction(expect_final=False))
            self.internal_pipeline.add_action(WaitDevicePathAction(self.usb_mass_device))
            self.internal_pipeline.add_action(FlashUBootUMSAction(self.usb_mass_device))
            self.internal_pipeline.add_action(ResetDevice())
        else:
            self.internal_pipeline.add_action(BootloaderCommandsAction())
        if self.has_prompts(parameters):
            self.internal_pipeline.add_action(AutoLoginAction())
            if self.test_has_shell(parameters):
                self.internal_pipeline.add_action(ExpectShellSession())
                if 'transfer_overlay' in parameters:
                    self.internal_pipeline.add_action(OverlayUnpack())
                self.internal_pipeline.add_action(ExportDeviceEnvironment())

    def validate(self):
        super(UBootRetry, self).validate()
        self.set_namespace_data(
            action=self.name,
            label='bootloader_prompt',
            key='prompt',
            value=self.job.device['actions']['boot']['methods']['u-boot']['parameters']['bootloader_prompt']
        )

    def run(self, connection, max_end_time, args=None):
        connection = super(UBootRetry, self).run(connection, max_end_time, args)
        self.set_namespace_data(action='shared', label='shared', key='connection', value=connection)
        return connection class ResetDevice(Action):
    """
    Used within a RetryAction - first tries 'reboot' then
    tries PDU.
    """

    name = "reset-device"
    description = "reboot or power-cycle the device"
    summary = "reboot the device"

    def populate(self, parameters):
        self.internal_pipeline = Pipeline(parent=self, job=self.job, parameters=parameters)
        if self.job.device.hard_reset_command:
            self.internal_pipeline.add_action(PDUReboot())
        else:
            self.internal_pipeline.add_action(SendRebootCommands()) class PDUReboot(Action):
    """
    Issues the PDU power cycle command on the dispatcher
    Raises InfrastructureError if either the command fails
    (pdu client reports error) or if the connection times out
    waiting for the device to reset.
    It is an error for a device to fail to reboot after a
    soft reboot and a failed hard reset.
    """

    name = "pdu-reboot"
    description = "issue commands to a PDU to power cycle a device"
    summary = "hard reboot using PDU"
    timeout_exception = InfrastructureError

    def __init__(self):
        super(PDUReboot, self).__init__()
        self.command = None

    def run(self, connection, max_end_time, args=None):
        connection = super(PDUReboot, self).run(connection, max_end_time, args)
        if not self.job.device.hard_reset_command:
            raise InfrastructureError("Hard reset required but not defined.")
        command = self.job.device.hard_reset_command
        if not isinstance(command, list):
            command = [command]
        for cmd in command:
            if not self.run_command(cmd.split(' '), allow_silent=True):
                raise InfrastructureError("%s failed" % cmd)
        self.results = {'status': 'success'}
        return connection def infrastructure_error(path):
    """
    Extends which into a check which sets default messages for Action validation,
    without needing to raise an Exception (which is slow).
    Use for quick checks on whether essential tools are installed and usable.
    """
    exefile = _which_check(path, match=os.path.isfile)
    if not exefile:
        return "Cannot find command '%s' in $PATH" % path
    # is the infrastructure call safe to make?
    if exefile and os.stat(exefile).st_mode & S_IXUSR != S_IXUSR:
        return "%s is not executable" % exefile
    return None 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 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 VCSHelper(object):
class BzrHelper(VCSHelper):
class GitHelper(VCSHelper):
class TarHelper(VCSHelper):
class URLHelper(VCSHelper):
    def __init__(self, url):
    def clone(self, dest_path, revision=None, branch=None):
    def __init__(self, url):
    def clone(self, dest_path, revision=None, branch=None):
    def __init__(self, url):
    def clone(self, dest_path, shallow=False, revision=None, branch=None, history=True):
    def __init__(self, url):
    def __init__(self, url): def add_to_kickstart(preseedfile, extra_command):
def add_late_command(preseedfile, extra_command):
def endsin(lines, endstring):
import os
import shutil
import subprocess
import tempfile
import unittest
from lava_dispatcher.utils.filesystem import mkdtemp
from lava_dispatcher.test.test_uboot import UBootFactory, StdoutTestCase
from lava_dispatcher.actions.boot.u_boot import UBootAction, UBootRetry
from lava_dispatcher.power import ResetDevice, PDUReboot
from lava_dispatcher.test.utils import infrastructure_error
from lava_dispatcher.action import InfrastructureError, Action
from lava_dispatcher.utils import vcs, installers
        from lava_dispatcher.actions.deploy import strategies  # pylint: disable=unused-variable
        from lava_dispatcher.actions.boot import strategies  # pylint: disable=reimported
        from lava_dispatcher.actions.test import strategies  # pylint: disable=reimported
# Copyright (C) 2014 Linaro Limited
#
# Author: Remi Duraffort remi.duraffort@linaro.org>
#
# This file is part of LAVA Dispatcher.
#
# LAVA Dispatcher is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# LAVA Dispatcher is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along
# with this program; if not, see <http://www.gnu.org/licenses>.




class TestGit(StdoutTestCase):  # pylint: disable=too-many-public-methods

    def setUp(self):
        super(TestGit, self).setUp()
        self.cwd = os.getcwd()

        # Go into a temp dirctory
        self.tmpdir = tempfile.mkdtemp()
        os.chdir(self.tmpdir)

        # Create a Git repository with two commits
        subprocess.check_output(['git', 'init', 'git'])
        os.chdir('git')
        with open('test.txt', 'w') as testfile:
            testfile.write("Some data")
        subprocess.check_output(['git', 'add', 'test.txt'])
        subprocess.check_output(['git', 'commit', 'test.txt', '-m', 'First commit'],
                                env={'GIT_COMMITTER_DATE': 'Fri Oct 24 14:40:36 CEST 2014',
                                     'GIT_AUTHOR_DATE': 'Fri Oct 24 14:40:36 CEST 2014',
                                     'GIT_AUTHOR_NAME': 'Foo Bar',
                                     'GIT_AUTHOR_EMAIL': 'foo@example.com',
                                     'GIT_COMMITTER_NAME': 'Foo Bar',
                                     'GIT_COMMITTER_EMAIL': 'foo@example.com'})
        with open('second.txt', 'w') as datafile:
            datafile.write("Some more data")
        subprocess.check_output(['git', 'add', 'second.txt'])
        subprocess.check_output(['git', 'commit', 'second.txt', '-m', 'Second commit'],
                                env={'GIT_COMMITTER_DATE': 'Fri Oct 24 14:40:38 CEST 2014',
                                     'GIT_AUTHOR_DATE': 'Fri Oct 24 14:40:38 CEST 2014',
                                     'GIT_AUTHOR_NAME': 'Foo Bar',
                                     'GIT_AUTHOR_EMAIL': 'foo@example.com',
                                     'GIT_COMMITTER_NAME': 'Foo Bar',
                                     'GIT_COMMITTER_EMAIL': 'foo@example.com'})

        subprocess.check_output(['git', 'checkout', '-q', '-b', 'testing'])
        with open('third.txt', 'w') as datafile:
            datafile.write("333")
        subprocess.check_output(['git', 'add', 'third.txt'])
        subprocess.check_output(['git', 'commit', 'third.txt', '-m', 'Third commit'],
                                env={'GIT_COMMITTER_DATE': 'Thu Sep  1 10:14:29 CEST 2016',
                                     'GIT_AUTHOR_DATE': 'Thu Sep  1 10:14:29 CEST 2016',
                                     'GIT_AUTHOR_NAME': 'Foo Bar',
                                     'GIT_AUTHOR_EMAIL': 'foo@example.com',
                                     'GIT_COMMITTER_NAME': 'Foo Bar',
                                     'GIT_COMMITTER_EMAIL': 'foo@example.com'})

        subprocess.check_output(['git', 'checkout', '-q', 'master'])

        # Go into the tempdir
        os.chdir('..')

    def tearDown(self):
        os.chdir(self.cwd)
        # Remove everything
        shutil.rmtree(self.tmpdir)

    def test_simple_clone(self):
        git = vcs.GitHelper('git')
        self.assertEqual(git.clone('git.clone1'), 'a7af835862da0e0592eeeac901b90e8de2cf5b67')
        self.assertEqual(git.clone('git.clone2'), 'a7af835862da0e0592eeeac901b90e8de2cf5b67')
        self.assertEqual(git.clone('git.clone3'), 'a7af835862da0e0592eeeac901b90e8de2cf5b67')

    def test_clone_at_head(self):
        git = vcs.GitHelper('git')
        self.assertEqual(git.clone('git.clone1', revision='a7af835862da0e0592eeeac901b90e8de2cf5b67'), 'a7af835862da0e0592eeeac901b90e8de2cf5b67')

    def test_clone_at_head_1(self):
        git = vcs.GitHelper('git')
        self.assertEqual(git.clone('git.clone1', revision='2f83e6d8189025e356a9563b8d78bdc8e2e9a3ed'), '2f83e6d8189025e356a9563b8d78bdc8e2e9a3ed')
        self.assertEqual(git.clone('git.clone2', revision='2f83e6d8189025e356a9563b8d78bdc8e2e9a3ed'), '2f83e6d8189025e356a9563b8d78bdc8e2e9a3ed')

    def test_non_existing_git(self):
        git = vcs.GitHelper('does_not_exists')
        self.assertRaises(InfrastructureError, git.clone, ('foo.bar'))

    def test_existing_destination(self):
        git = vcs.GitHelper('git')
        self.assertEqual(git.clone('git.clone1'), 'a7af835862da0e0592eeeac901b90e8de2cf5b67')
        self.assertRaises(InfrastructureError, git.clone, ('git.clone1'))
        self.assertRaises(InfrastructureError, git.clone, ('git'))

    def test_invalid_commit(self):
        git = vcs.GitHelper('git')
        self.assertRaises(InfrastructureError, git.clone, 'foo.bar', True,
                          'badhash')

    def test_branch(self):
        git = vcs.GitHelper('git')
        self.assertEqual(git.clone('git.clone1', branch='testing'), 'f2589a1b7f0cfc30ad6303433ba4d5db1a542c2d')
        self.assertTrue(os.path.exists(os.path.join(self.tmpdir, "git.clone1", ".git")))

    def test_no_history(self):
        git = vcs.GitHelper('git')
        self.assertEqual(git.clone('git.clone1', history=False), 'a7af835862da0e0592eeeac901b90e8de2cf5b67')
        self.assertFalse(os.path.exists(os.path.join(self.tmpdir, "git.clone1", ".git")))



@unittest.skipIf(infrastructure_error('bzr'), "bzr not installed")