Source code for autotest.client.harness_autoserv

import ConfigParser
import logging
import os

from autotest.client import harness
from autotest.client.shared import autotemp, base_packages, error
from autotest.client.shared.settings import settings

[docs]class harness_autoserv(harness.harness): """ The server harness for running from autoserv Properties: job The job object for this job """ def __init__(self, job, harness_args): """ job The job object for this job """ super(harness_autoserv, self).__init__(job) self.status = os.fdopen(3, 'w', 0) # If a bug on the client run code prevents global_config.ini # from being copied to the client machine, the client will run # without a global config, relying only on the defaults of the # config items. To avoid that happening silently, the check below # was written. try: settings.get_section_values(["CLIENT", "COMMON"]) except ConfigParser.NoSectionError: logging.error("Empty CLIENT or COMMON configuration session. " "global_config.ini missing. This probably means " "a bug on the server code. Please verify.")
[docs] def run_start(self): # set up the package fetcher for direct-from-autoserv fetches fetcher = AutoservFetcher(self.job.pkgmgr, self) self.job.pkgmgr.add_repository(fetcher)
def _send_and_wait(self, title, *args): """Send a message to the autoserv and wait for it to signal completion. :param title: An alphanumeric string to title the message. :param *args: Additional arbitrary alphanumeric arguments to pass to the server. """ # create a named pipe for us to receive a signal on fifo_dir = autotemp.tempdir(suffix='-fifo', unique_id='harness', dir=self.job.tmpdir) try: fifo_path = os.path.join(, 'autoserv.fifo') os.mkfifo(fifo_path) # send signal to the server as title[:args]:path msg = ':'.join([title] + list(args) + [fifo_path]) + '\n' self.status.write(msg) # wait for the server to signal back to us fifo = open(fifo_path) fifo.close() finally: fifo_dir.clean()
[docs] def run_test_complete(self): """A test run by this job is complete, signal it to autoserv and wait for it to signal to continue""" self._send_and_wait('AUTOTEST_TEST_COMPLETE')
[docs] def test_status(self, status, tag): """A test within this job is completing""" for line in status.split('\n'): # sent status messages with AUTOTEST_STATUS:tag:message msg = 'AUTOTEST_STATUS:%s:%s\n' msg %= (tag, line) self.status.write(msg)
[docs] def fetch_package(self, pkg_name, dest_path): """Request a package from the remote autoserv. :param pkg_name: The name of the package, as generally used by the client.shared.packages infrastructure. :param dest_path: The path the package should be copied to. """ self._send_and_wait('AUTOTEST_FETCH_PACKAGE', pkg_name, dest_path)
[docs]class AutoservFetcher(base_packages.RepositoryFetcher): def __init__(self, package_manager, job_harness): super(AutoservFetcher, self).__init__(package_manager=package_manager, repository_url="autoserv://") self.job_harness = job_harness
[docs] def fetch_pkg_file(self, filename, dest_path): if os.path.exists(dest_path): os.remove(dest_path)'Fetching %s from autoserv to %s', filename, dest_path) self.job_harness.fetch_package(filename, dest_path) if os.path.exists(dest_path): logging.debug('Successfully fetched %s from autoserv', filename) else: raise error.PackageFetchError('%s not fetched from autoserv' % filename)