# Ubuntu Testing Automation Harness
# Copyright 2012 Canonical Ltd.
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, 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/>.
"""Provide code to actually run the tests."""
import datetime
import jsonschema
import os
import shutil
import stat
import subprocess
import urllib
from utah import logger
from utah.client import exceptions
from utah.client.probe import PROBES, probes
from utah.client.common import (
CMD_TC_REBOOT,
DATE_FORMAT,
DEFAULT_STATE_FILE,
DEFAULT_TSLIST,
DefaultValidator,
MASTER_RUNLIST,
ReturnCodes,
UTAH_DIR,
chdir,
make_result,
mkdir,
parse_yaml_file,
)
from utah.client.package_info import (
InstalledPackages,
DeltaPackages,
)
from utah.client.state_agent import StateAgentYAML
from utah.client.testsuite import TestSuite, DynamicTestSuite
from utah.client.vcs import create_vcs_handler
from utah.retry import retry
from utah.timeout import timeout
RC_LOCAL = '/etc/rc.local'
RC_LOCAL_BACKUP = '{}-utah.bak'.format(RC_LOCAL)
# XXX: need to output to the same file that was supplied on the original
# run.
rc_local_content = """#!/bin/sh
/usr/bin/utah --resume -o {output} -r {runlist}
"""
[docs]class Runner(object):
"""Provide The main runner class.
Parses a master runlist, builds a list of TestSuites, and runs them.
"""
status = "NOTRUN"
utah_exec_path = '/usr/bin/utah'
MASTER_RUNLIST_SCHEMA = {
'type': 'object',
'properties': {
'testsuites': {
'type': 'array',
'items': {
'type': 'object',
'oneOf': [
{'$ref': '#/definitions/testsuite_fetch'},
{'$ref': '#/definitions/testsuite_file'},
],
},
'minItems': 1,
},
'battery_measurements': {
'type': 'boolean',
'default': False,
'description': 'DEPRECATED in favor of "probes"',
},
'probes': {
'type': 'array',
'items': {'enum': PROBES.keys()},
'default': [],
'description': 'A list of probes to enable for the runlist.',
},
'timeout': {
'type': 'integer',
'minimum': 0,
},
'repeat_count': {
'description': '''By default this does nothing and the suite
is run 1 time. However, if repeat_count != 0, it will be
repeated X times. So a repeat_count=2 means the suite will
be run 3 times.
''',
'type': 'integer',
'minimum': 0,
'default': 0,
},
},
'required': ['testsuites'],
'additionalProperties': False,
'definitions': {
'testsuite_fetch': {
'type': 'object',
'properties': {
'name': {'type': 'string'},
'fetch_method': {
'type': 'string',
'enum': ['dev', 'bzr', 'bzr-export', 'git'],
},
'fetch_location': {'type': 'string'},
'include_tests': {
'type': 'array',
'items': {'type': 'string'},
'minItems': 1,
},
'exclude_tests': {
'type': 'array',
'items': {'type': 'string'},
'minItems': 1,
},
},
'required': ['name', 'fetch_method', 'fetch_location'],
'additionalProperties': False,
},
'testsuite_file': {
'type': 'object',
'properties': {
'include': {'type': 'string'},
},
'required': ['include'],
'additionalProperties': False,
},
}
}
def __init__(self, install_type, runlist=None, result=None,
testdir=UTAH_DIR, state_agent=None,
resume=False, old_results=None, output=None, repeat_count=0):
# Runlist URL passed through the command line
self.testdir = testdir
self.revision = "Unknown"
self.output = (output or
os.path.join('/', 'var', 'lib', 'utah', 'utah.out'))
self.testsuitedir = os.path.join(testdir, 'testsuites')
# The testdir must exist, XXX: perhaps we can create it, needs
# discussion.
if not os.path.exists(testdir):
raise exceptions.BadDir(testdir)
mkdir(self.testsuitedir)
# If no runlist is supplied look for one in the testdir
if runlist is None:
runlist = os.path.join(testdir, 'master.run')
if not os.path.isfile(runlist):
raise exceptions.MissingFile(
'Runlist {} does not exist'.format(runlist))
self.master_runlist = runlist
self.suites = []
self.result = result
if old_results:
self.result.load(old_results)
self.state_file = DEFAULT_STATE_FILE
self.fetched_suites = {}
self.timeout = None
self.install_type = install_type
self.repeat_count = repeat_count
self.errors = 0
self.passes = 0
self.failures = 0
self.fetch_errors = 0
self.state_agent = state_agent or StateAgentYAML()
# Cleanup the state file if this is supposed to be a fresh run.
if not resume:
self.state_agent.clean()
else:
self.reset_rc_local()
self.load_state()
chdir(self.testsuitedir)
# needs to be run prior to process_master_runlist since that's where
# test suites are fetched.
self.fetched_suites = self.get_fetched_suites()
self.process_master_runlist(resume=resume)
self.result.runlist = self.master_runlist
self.result.install_type = self.install_type
self.result.name = self.name
[docs] def backup_rc_local(self):
"""Backup /etc/rc.local if it exists."""
# Ignore permission denied errors since we only care if a reboot is
# pending whether or not we can write to RC_LOCAL(_BACKUP).
try:
os.rename(RC_LOCAL, RC_LOCAL_BACKUP)
except OSError as e:
if e.errno != 13:
raise exceptions.UTAHClientError(
'Failed to backup rc.local: {}'.format(e))
[docs] def reset_rc_local(self):
"""Restore /etc/rc.local if there is a backup, remove it otherwise."""
# Ignore permission denied errors since we only care if a reboot is
# pending whether or not we can write to RC_LOCAL(_BACKUP).
try:
if os.path.exists(RC_LOCAL_BACKUP):
os.rename(RC_LOCAL_BACKUP, RC_LOCAL)
elif os.path.exists(RC_LOCAL):
os.remove(RC_LOCAL)
except OSError as e:
if e.errno != 13:
raise exceptions.UTAHClientError(
'Failed to backup rc.local: {}'.format(e))
[docs] def setup_rc_local(self, rc_local=RC_LOCAL, runlist=None):
"""Setup /etc/rc.local to kick-off a --resume on successful boot."""
runlist = runlist or self.master_runlist or MASTER_RUNLIST
try:
with open(rc_local, 'w') as fp:
fp.write(rc_local_content.format(runlist=runlist,
output=self.output))
os.fchmod(fp.fileno(),
stat.S_IRWXU | stat.S_IROTH | stat.S_IRGRP)
except (IOError, OSError) as err:
raise exceptions.UTAHClientError(
'Error setting up rc.local: {}'.format(err))
[docs] def process_results(self):
"""Add stats to results and process them.
:returns: A return code based on the test status.
:rtype: int
"""
# Add stats to the results
self.result.failures = self.failures
self.result.passes = self.passes
self.result.errors = self.errors
self.result.fetch_errors = self.fetch_errors
# process the results
self.result.result()
return self.returncode()
def _run_before(self):
"""Helper function with things to be done *before* a test run."""
# Gather installed packages and their version before running any test
self.start_packages = InstalledPackages()
self.result.probe_data = probes.start()
def _run_after(self):
"""Helper function with things to be done *after* a test run."""
self.result.probe_data = probes.stop(self.result.probe_data)
# Gather installed packages and their version after running all test
end_packages = InstalledPackages()
delta_packages = DeltaPackages(self.start_packages, end_packages)
self.result.packages = dict(self.start_packages)
self.result.delta_packages = dict(delta_packages)
[docs] def run(self, iteration=0):
"""Run the test suites we've parsed.
:returns: The result of process_results, which is a return code.
:rtype: int
.. seealso:: :meth:`process_results`
"""
self._run_before()
# Return value to indicate whether processing of a Runner should
# continue. This is to avoid a shutdown race on reboot cases.
keep_going = True
self.status = "RUN"
for suite in self.suites:
chdir(self.testsuitedir)
if iteration or not suite.is_done():
keep_going = suite.run(iteration != 0)
if not keep_going:
# Return success despite of the results
# to let server know that it has to check the state file
# to know the final result after the reboot
logger.log('REBOOT: Test case requested reboot')
return ReturnCodes.REBOOT
self.errors += suite.errors
self.passes += suite.passes
self.failures += suite.failures
if self.repeat_count:
self.repeat_count -= 1
self.status = "DONE"
self.save_state()
self._run_after()
[docs] def add_suite(self, suite):
"""Add a test suite to run."""
self.suites.append(suite)
[docs] def get_fetched_suites(self):
"""Return a list of fetched suites from the state_agent."""
state = self.state_agent.load_state()
fetched_suites = []
if state and 'fetched_suites' in state:
fetched_suites = state['fetched_suites']
return fetched_suites
[docs] def load_state(self):
"""Load the state saved by a previous partial run (i.e., a reboot)."""
state = self.state_agent.load_state()
self.master_runlist = state['master_runlist']
self.status = state['status']
self.repeat_count = state['repeat_count']
self.suites = []
for state_suite in state['suites']:
suite = TestSuite(state_suite['name'], self)
suite.load_state(state_suite)
self.suites.append(suite)
self.fetched_suites = state['fetched_suites']
[docs] def save_state(self):
"""Save the list of tests we are to run and whether we've run them.
:returns: state of currently run tests
:rtype: dict
"""
self.backup_runlist = os.path.join(self.testdir, 'master.run-reboot')
if (os.path.exists(
self.master_runlist)
and self.master_runlist != self.backup_runlist):
try:
shutil.copyfile(self.master_runlist, self.backup_runlist)
except OSError as err:
raise exceptions.UTAHClientError(
'Failed to backup master runlist from {} to {}: {}'
.format(self.master_runlist, self.backup_runlist, err))
state = {
'master_runlist': self.backup_runlist,
'original_runlist': self.master_runlist,
'status': self.status,
'passes': self.passes,
'failures': self.failures,
'errors': self.errors,
'fetch_errors': self.fetch_errors,
'suites': [],
'fetched_suites': self.fetched_suites,
'testdir': self.testdir,
'repeat_count': self.repeat_count,
}
for suite in self.suites:
state['suites'].append(suite.save_state())
# hand the state dictionary off to the StateAgent to process
self.state_agent.save_state(state)
return state
[docs] def count_suites(self):
"""Return the number of test suites in the runner."""
return len(self.suites)
[docs] def count_tests(self):
"""Return the number of test cases in the runner."""
tests = 0
for suite in self.suites:
tests += suite.count_tests()
return tests
def _fetch_suite(self, runlist, suite):
name = suite['name']
logger.log('Fetching testsuite: {}'.format(name))
vcs_handler = create_vcs_handler(
suite['fetch_method'],
suite['fetch_location'],
name,
runlist,
)
res = timeout(60, retry, vcs_handler.get, name)
self.result.add_result(res)
fetch_success = self.result.status == 'PASS'
if fetch_success:
rev_dir = os.path.join(name, name)
res = vcs_handler.revision(directory=rev_dir)
self.result.add_result(res)
# If we're unable to get the revision number indicate a
# fetch failure but allow the test runs to proceed.
if self.result.status != 'PASS':
self.result.status = 'PASS'
return fetch_success
@staticmethod
def _fetch_and_parse(runlist):
try:
local_filename = urllib.urlretrieve(runlist)[0]
except IOError as err:
raise exceptions.MissingFile(
'Error when downloading {}: {}'.format(runlist, err))
data = parse_yaml_file(local_filename)
validator = DefaultValidator(Runner.MASTER_RUNLIST_SCHEMA)
try:
validator.validate(data)
if data['battery_measurements']:
if 'battery' not in data['probes']:
data['probes'].append('battery')
probes.enable_probes(data['probes'])
except jsonschema.ValidationError as exception:
raise exceptions.ValidationError(
'Master runlist failed to validate: {!r}\n'
'Detailed information: {}'
.format(local_filename, exception))
return data
@staticmethod
def _clean_suite(name):
# convert to absolute name to make troubleshooting a little easier
name = os.path.abspath(name)
try:
shutil.rmtree(name)
except OSError as e:
raise exceptions.UTAHClientError(
'Error removing the testsuite {}: {}'.format(name, e))
def _add_suite(self, runlist, suite):
name = suite['name']
if name not in self.fetched_suites:
if self._fetch_suite(runlist, suite):
self.fetched_suites.append(name)
auto = os.path.join(name, name, 'tslist.auto')
if os.path.exists(auto):
ts = DynamicTestSuite(name, self, 'tslist.auto')
else:
suite_runlist = suite.get('runlist', DEFAULT_TSLIST)
inc = suite.get('include_tests', None)
exc = suite.get('exclude_tests', None)
ts = TestSuite(name, self, suite_runlist, inc, exc)
self.add_suite(ts)
else:
self.fetch_errors += 1
[docs] def process_master_runlist(self, runlist=None, resume=False):
"""Parse a master runlist and build a list of suites from the data.
:param runlist: URL pointing to a runlist
:type rulist: string
:param resume: Continue previous execution
:type resume: boolean
"""
runlist = runlist or self.master_runlist
data = self._fetch_and_parse(runlist)
if 'timeout' in data:
self.timeout = int(data['timeout'])
self.name = data.get('name', 'unnamed')
self.probes = data['probes']
self.repeat_count = data['repeat_count']
seen = []
orig_dir = os.getcwd()
suites = data['testsuites']
suites = data['testsuites'] if 'testsuites' in data else data
for suite in suites:
# Allow the inclusion of other master.run files
if 'include' in suite:
self.process_master_runlist(suite['include'])
continue
chdir(orig_dir)
name = suite['name']
if name in seen:
raise exceptions.BadMasterRunlist(
"{} duplicated in runlist".format(name))
# Fetch the testsuite. On resume don't remove the testsuite
# directory.
if not resume and os.path.exists(name):
self._clean_suite(name)
mkdir(name)
self._add_suite(runlist, suite)
[docs] def get_next_suite(self):
"""Return the next suite to be run.
Mainly used for debugging.
"""
suite = None
for s in self.suites:
if not s.is_done():
suite = s
break
return suite
[docs] def get_next_test(self):
"""Return the next test to be run.
Mainly used for debugging.
"""
return self.get_next_suite().get_next_test()
def _autorun_file(self):
"""Determine if this was launched via the autorun feature.
:return: the autorun file or None
"""
r = os.listdir('/var/cache/utah/autorun/inprogress')
if len(r) == 1 and 'run-utah' in r[0]:
return r[0]
return None
[docs] def reboot(self):
"""Reboot the machine.
Save state, setup /etc/rc.local, and shutdown.
"""
# Create fake result for logging purposes
command = 'shutdown -r now'
start_time = datetime.datetime.now()
cmd_result = make_result(
command=command,
retcode=0,
start_time=start_time.strftime(DATE_FORMAT),
time_delta=str(datetime.timedelta()),
cmd_type=CMD_TC_REBOOT,
user='root',
)
self.result.add_result(cmd_result)
self.result.result()
self.save_state()
f = self._autorun_file()
if f:
logger.log('restarting client via autostart on reboot')
self.setup_rc_local(runlist=self.backup_runlist, rc_local=f)
os.rename(f, '/etc/utah/autorun/01_run-utah')
else:
logger.log('restarting client via rc.local on reboot')
self.backup_rc_local()
self.setup_rc_local(runlist=self.backup_runlist)
try:
# let the shutdown run in the background in a few seconds to allow
# the client enough time to properly send an exit code back to
# the server
subprocess.call('(sleep 3; shutdown -r now)&', shell=True)
except OSError as err:
raise exceptions.UTAHClientError('Failed to reboot: {}'
.format(err))
# End of execution
[docs] def returncode(self):
"""Provide return code based on test status."""
if self.errors > 0 or self.fetch_errors > 0:
logger.log_error('ERROR: Errors encountered during testing')
return ReturnCodes.ERROR
elif self.failures > 0:
logger.log_error('FAIL: Not all tests passed')
return ReturnCodes.FAIL
else:
logger.log('PASS: All tests passed')
return ReturnCodes.PASS