server/site_autotest.py

# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import logging
import os
from autotest_lib.client.common_lib import error, global_config
from autotest_lib.client.common_lib.cros import dev_server
from autotest_lib.server import installable_object, autoserv_parser
from autotest_lib.server.cros.dynamic_suite import tools
from autotest_lib.server.cros.dynamic_suite.constants import JOB_REPO_URL


_CONFIG = global_config.global_config
_PARSER = autoserv_parser.autoserv_parser


class SiteAutotest(installable_object.InstallableObject):

    def get(self, location=None):
        if not location:
            location = os.path.join(self.serverdir, '../client')
            location = os.path.abspath(location)
        installable_object.InstallableObject.get(self, location)
        self.got = True


    def _get_fetch_location_from_host_attribute(self):
        """Get repo to use for packages from host attribute, if possible.

        Hosts are tagged with an attribute containing the URL
        from which to source packages when running a test on that host.
        If self.host is set, attempt to look this attribute up by calling out
        to the AFE.

        @returns value of the 'job_repo_url' host attribute, if present.
        """
        try:
            from autotest_lib.server import frontend
            if self.host:
                afe = frontend.AFE(debug=False)
                hosts = afe.get_hosts(hostname=self.host.hostname)
                if hosts and JOB_REPO_URL in hosts[0].attributes:
                    return hosts[0].attributes[JOB_REPO_URL]
                logging.warning("No %s for %s", JOB_REPO_URL, self.host)
        except ImportError:
            logging.warning('Not attempting to look for %s', JOB_REPO_URL)
            pass
        return None


    def get_fetch_location(self):
        """Generate list of locations where autotest can look for packages.

        Old n' busted: Autotest packages are always stored at a URL that can
        be derived from the one passed via the voodoo magic --image argument.
        New hotness: Hosts are tagged with an attribute containing the URL
        from which to source packages when running a test on that host.

        @returns the list of candidate locations to check for packages.
        """
        repos = super(SiteAutotest, self).get_fetch_location()

        if _PARSER.options.image:
            image_opt = _PARSER.options.image
            if image_opt.startswith('http://'):
                # A devserver HTTP url was specified, set that as the repo_url.
                repos.append(image_opt.replace(
                    'update', 'static/archive').rstrip('/') + '/autotest')
            else:
                # An image_name like stumpy-release/R27-3437.0.0 was specified,
                # set this as the repo_url for the host. If an AFE is not being
                # run, this will ensure that the installed build uses the
                # associated artifacts for the test specified when running
                # autoserv with --image. However, any subsequent tests run on
                # the host will no longer have the context of the image option
                # and will revert back to utilizing test code/artifacts that are
                # currently present in the users source checkout.
                devserver_url = dev_server.ImageServer.resolve(image_opt).url()
                repo_url = tools.get_package_url(devserver_url, image_opt)
                repos.append(repo_url)
        else:
            # No --image option was specified, look for the repo url via
            # the host attribute. If we are not running with a full AFE
            # autoserv will fall back to serving packages itself from whatever
            # source version it is sync'd to rather than using the proper
            # artifacts for the build on the host.
            found_repo = self._get_fetch_location_from_host_attribute()
            if found_repo is not None:
                # Add our new repo to the end, the package manager will
                # later reverse the list of repositories resulting in ours
                # being first
                repos.append(found_repo)

        return repos


    def install(self, host=None, autodir=None):
        """Install autotest.  If |host| is not None, stores it in |self.host|.

        @param host A Host instance on which autotest will be installed
        @param autodir Location on the remote host to install to
        """
        if host:
            self.host = host

        super(SiteAutotest, self).install(host=host, autodir=autodir)


    def run_static_method(self, module, method, results_dir='.', host=None,
                          *args):
        """Runs a non-instance method with |args| from |module| on the client.

        This method runs a static/class/module autotest method on the client.
        For example:
          run_static_method("autotest_lib.client.cros.cros_ui", "reboot")

        Will run autotest_lib.client.cros.cros_ui.reboot() on the client.

        @param module: module name as you would refer to it when importing in a
            control file. e.g. autotest_lib.client.common_lib.module_name.
        @param method: the method you want to call.
        @param results_dir: A str path where the results should be stored
            on the local filesystem.
        @param host: A Host instance on which the control file should
            be run.
        @param args: args to pass to the method.
        """
        control = "\n".join(["import %s" % module,
                             "%s.%s(%s)\n" % (module, method,
                                              ','.join(map(repr, args)))])
        self.run(control, results_dir=results_dir, host=host)


class SiteClientLogger(object):
    """Overrides default client logger to allow for using a local package cache.
    """

    def _process_line(self, line):
        """Returns the package checksum file if it exists."""
        logging.debug(line)
        fetch_package_match = self.fetch_package_parser.search(line)
        if fetch_package_match:
            pkg_name, dest_path, fifo_path = fetch_package_match.groups()
            serve_packages = _CONFIG.get_config_value(
                "PACKAGES", "serve_packages_from_autoserv", type=bool)
            if serve_packages and pkg_name == 'packages.checksum':
                try:
                    checksum_file = os.path.join(
                        self.job.pkgmgr.pkgmgr_dir, 'packages', pkg_name)
                    if os.path.exists(checksum_file):
                        self.host.send_file(checksum_file, dest_path)
                except error.AutoservRunError:
                    msg = "Package checksum file not found, continuing anyway"
                    logging.exception(msg)

                try:
                    # When fetching a package, the client expects to be
                    # notified when the fetching is complete. Autotest
                    # does this pushing a B to a fifo queue to the client.
                    self.host.run("echo B > %s" % fifo_path)
                except error.AutoservRunError:
                    msg = "Checksum installation failed, continuing anyway"
                    logging.exception(msg)
                finally:
                    return

        # Fall through to process the line using the default method.
        super(SiteClientLogger, self)._process_line(line)


    def _send_tarball(self, pkg_name, remote_dest):
        """Uses tarballs in package manager by default."""
        try:
            server_package = os.path.join(self.job.pkgmgr.pkgmgr_dir,
                                          'packages', pkg_name)
            if os.path.exists(server_package):
              self.host.send_file(server_package, remote_dest)
              return

        except error.AutoservRunError:
            msg = ("Package %s could not be sent from the package cache." %
                   pkg_name)
            logging.exception(msg)

        # Fall through to send tarball the default method.
        super(SiteClientLogger, self)._send_tarball(pkg_name, remote_dest)


class _SiteRun(object):
    pass