blob: 0f9f2a5abf60e1c037948bb6d2c8628cf673db94 [file] [log] [blame]
# 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
import tempfile
import urllib2
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) = 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 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.
from autotest_lib.server import frontend
afe = frontend.AFE(debug=False)
hosts = afe.get_hosts(
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,
except (ImportError, urllib2.URLError):
logging.warning('Not attempting to look for %s', JOB_REPO_URL)
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.
'update', 'static').rstrip('/') + '/autotest')
# 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)
# 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
return repos
def install(self, host=None, autodir=None):
"""Install autotest. If |host| is not None, stores it in ||.
@param host A Host instance on which autotest will be installed
@param autodir Location on the remote host to install to
if host: = host
super(SiteAutotest, self).install(host=host, autodir=autodir)
def _install(self, host=None, autodir=None, use_autoserv=True,
Install autotest. If get() was not called previously, an
attempt will be made to install from the autotest svn
@param host A Host instance on which autotest will be installed
@param autodir Location on the remote host to install to
@param use_autoserv Enable install modes that depend on the client
running with the autoserv harness
@param use_packaging Enable install modes that use the packaging system
@exception AutoservError if a tarball was not specified and
the target host does not have svn installed in its path
# TODO(milleral):
super(SiteAutotest, self)._install(host, autodir, use_autoserv,
# Send over the most recent global_config.ini after installation if one
# is available.
# This code is a bit duplicated from
# _BaseRun._create_client_config_file, but oh well.
if self.installed and self.source_material:'Installing updated global_config.ini.')
destination = os.path.join(,
with tempfile.NamedTemporaryFile() as client_config:
config = global_config.global_config
client_section = config.get_section_values('CLIENT')
client_config.flush(), destination)
def run_static_method(self, module, method, results_dir='.', host=None,
"""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)))]), 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."""
fetch_package_match =
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':
checksum_file = os.path.join(
self.job.pkgmgr.pkgmgr_dir, 'packages', pkg_name)
if os.path.exists(checksum_file):, dest_path)
except error.AutoservRunError:
msg = "Package checksum file not found, continuing anyway"
# 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."echo B > %s" % fifo_path)
except error.AutoservRunError:
msg = "Checksum installation failed, continuing anyway"
# 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."""
server_package = os.path.join(self.job.pkgmgr.pkgmgr_dir,
'packages', pkg_name)
if os.path.exists(server_package):, remote_dest)
except error.AutoservRunError:
msg = ("Package %s could not be sent from the package cache." %
# Fall through to send tarball the default method.
super(SiteClientLogger, self)._send_tarball(pkg_name, remote_dest)
class _SiteRun(object):