# pylint: disable-msg=C0111
|
# Copyright 2008 Google Inc. Released under the GPL v2
|
|
import warnings
|
with warnings.catch_warnings():
|
# The 'compiler' module is gone in Python 3.0. Let's not say
|
# so in every log file.
|
warnings.simplefilter("ignore", DeprecationWarning)
|
import compiler
|
import logging
|
import textwrap
|
import re
|
|
from autotest_lib.client.common_lib import enum
|
from autotest_lib.client.common_lib import global_config
|
from autotest_lib.client.common_lib import priorities
|
|
REQUIRED_VARS = set(['author', 'doc', 'name', 'time', 'test_type'])
|
OBSOLETE_VARS = set(['experimental'])
|
|
CONTROL_TYPE = enum.Enum('Server', 'Client', start_value=1)
|
CONTROL_TYPE_NAMES = enum.Enum(*CONTROL_TYPE.names, string_values=True)
|
|
_SUITE_ATTRIBUTE_PREFIX = 'suite:'
|
|
CONFIG = global_config.global_config
|
|
# Default maximum test result size in kB.
|
DEFAULT_MAX_RESULT_SIZE_KB = CONFIG.get_config_value(
|
'AUTOSERV', 'default_max_result_size_KB', type=int, default=20000)
|
|
|
class ControlVariableException(Exception):
|
pass
|
|
def _validate_control_file_fields(control_file_path, control_file_vars,
|
raise_warnings):
|
"""Validate the given set of variables from a control file.
|
|
@param control_file_path: string path of the control file these were
|
loaded from.
|
@param control_file_vars: dict of variables set in a control file.
|
@param raise_warnings: True iff we should raise on invalid variables.
|
|
"""
|
diff = REQUIRED_VARS - set(control_file_vars)
|
if diff:
|
warning = ('WARNING: Not all required control '
|
'variables were specified in %s. Please define '
|
'%s.') % (control_file_path, ', '.join(diff))
|
if raise_warnings:
|
raise ControlVariableException(warning)
|
print textwrap.wrap(warning, 80)
|
|
obsolete = OBSOLETE_VARS & set(control_file_vars)
|
if obsolete:
|
warning = ('WARNING: Obsolete variables were '
|
'specified in %s. Please remove '
|
'%s.') % (control_file_path, ', '.join(obsolete))
|
if raise_warnings:
|
raise ControlVariableException(warning)
|
print textwrap.wrap(warning, 80)
|
|
|
class ControlData(object):
|
# Available TIME settings in control file, the list must be in lower case
|
# and in ascending order, test running faster comes first.
|
TEST_TIME_LIST = ['fast', 'short', 'medium', 'long', 'lengthy']
|
TEST_TIME = enum.Enum(*TEST_TIME_LIST, string_values=False)
|
|
@staticmethod
|
def get_test_time_index(time):
|
"""
|
Get the order of estimated test time, based on the TIME setting in
|
Control file. Faster test gets a lower index number.
|
"""
|
try:
|
return ControlData.TEST_TIME.get_value(time.lower())
|
except AttributeError:
|
# Raise exception if time value is not a valid TIME setting.
|
error_msg = '%s is not a valid TIME.' % time
|
logging.error(error_msg)
|
raise ControlVariableException(error_msg)
|
|
|
def __init__(self, vars, path, raise_warnings=False):
|
# Defaults
|
self.path = path
|
self.dependencies = set()
|
# TODO(jrbarnette): This should be removed once outside
|
# code that uses can be changed.
|
self.experimental = False
|
self.run_verify = True
|
self.sync_count = 1
|
self.test_parameters = set()
|
self.test_category = ''
|
self.test_class = ''
|
self.job_retries = 0
|
# Default to require server-side package. Unless require_ssp is
|
# explicitly set to False, server-side package will be used for the
|
# job. This can be overridden by global config
|
# AUTOSERV/enable_ssp_container
|
self.require_ssp = None
|
self.attributes = set()
|
self.max_result_size_KB = DEFAULT_MAX_RESULT_SIZE_KB
|
self.priority = priorities.Priority.DEFAULT
|
self.fast = False
|
|
_validate_control_file_fields(self.path, vars, raise_warnings)
|
|
for key, val in vars.iteritems():
|
try:
|
self.set_attr(key, val, raise_warnings)
|
except Exception, e:
|
if raise_warnings:
|
raise
|
print 'WARNING: %s; skipping' % e
|
|
self._patch_up_suites_from_attributes()
|
|
|
@property
|
def suite_tag_parts(self):
|
"""Return the part strings of the test's suite tag."""
|
if hasattr(self, 'suite'):
|
return [part.strip() for part in self.suite.split(',')]
|
else:
|
return []
|
|
|
def set_attr(self, attr, val, raise_warnings=False):
|
attr = attr.lower()
|
try:
|
set_fn = getattr(self, 'set_%s' % attr)
|
set_fn(val)
|
except AttributeError:
|
# This must not be a variable we care about
|
pass
|
|
|
def _patch_up_suites_from_attributes(self):
|
"""Patch up the set of suites this test is part of.
|
|
Legacy builds will not have an appropriate ATTRIBUTES field set.
|
Take the union of suites specified via ATTRIBUTES and suites specified
|
via SUITE.
|
|
SUITE used to be its own variable, but now suites are taken only from
|
the attributes.
|
|
"""
|
|
suite_names = set()
|
# Extract any suites we know ourselves to be in based on the SUITE
|
# line. This line is deprecated, but control files in old builds will
|
# still have it.
|
if hasattr(self, 'suite'):
|
existing_suites = self.suite.split(',')
|
existing_suites = [name.strip() for name in existing_suites]
|
existing_suites = [name for name in existing_suites if name]
|
suite_names.update(existing_suites)
|
|
# Figure out if our attributes mention any suites.
|
for attribute in self.attributes:
|
if not attribute.startswith(_SUITE_ATTRIBUTE_PREFIX):
|
continue
|
suite_name = attribute[len(_SUITE_ATTRIBUTE_PREFIX):]
|
suite_names.add(suite_name)
|
|
# Rebuild the suite field if necessary.
|
if suite_names:
|
self.set_suite(','.join(sorted(list(suite_names))))
|
|
|
def _set_string(self, attr, val):
|
val = str(val)
|
setattr(self, attr, val)
|
|
|
def _set_option(self, attr, val, options):
|
val = str(val)
|
if val.lower() not in [x.lower() for x in options]:
|
raise ValueError("%s must be one of the following "
|
"options: %s" % (attr,
|
', '.join(options)))
|
setattr(self, attr, val)
|
|
|
def _set_bool(self, attr, val):
|
val = str(val).lower()
|
if val == "false":
|
val = False
|
elif val == "true":
|
val = True
|
else:
|
msg = "%s must be either true or false" % attr
|
raise ValueError(msg)
|
setattr(self, attr, val)
|
|
|
def _set_int(self, attr, val, min=None, max=None):
|
val = int(val)
|
if min is not None and min > val:
|
raise ValueError("%s is %d, which is below the "
|
"minimum of %d" % (attr, val, min))
|
if max is not None and max < val:
|
raise ValueError("%s is %d, which is above the "
|
"maximum of %d" % (attr, val, max))
|
setattr(self, attr, val)
|
|
|
def _set_set(self, attr, val):
|
val = str(val)
|
items = [x.strip() for x in val.split(',') if x.strip()]
|
setattr(self, attr, set(items))
|
|
|
def set_author(self, val):
|
self._set_string('author', val)
|
|
|
def set_dependencies(self, val):
|
self._set_set('dependencies', val)
|
|
|
def set_doc(self, val):
|
self._set_string('doc', val)
|
|
|
def set_name(self, val):
|
self._set_string('name', val)
|
|
|
def set_run_verify(self, val):
|
self._set_bool('run_verify', val)
|
|
|
def set_sync_count(self, val):
|
self._set_int('sync_count', val, min=1)
|
|
|
def set_suite(self, val):
|
self._set_string('suite', val)
|
|
|
def set_time(self, val):
|
self._set_option('time', val, ControlData.TEST_TIME_LIST)
|
|
|
def set_test_class(self, val):
|
self._set_string('test_class', val.lower())
|
|
|
def set_test_category(self, val):
|
self._set_string('test_category', val.lower())
|
|
|
def set_test_type(self, val):
|
self._set_option('test_type', val, list(CONTROL_TYPE.names))
|
|
|
def set_test_parameters(self, val):
|
self._set_set('test_parameters', val)
|
|
|
def set_job_retries(self, val):
|
self._set_int('job_retries', val)
|
|
|
def set_bug_template(self, val):
|
if type(val) == dict:
|
setattr(self, 'bug_template', val)
|
|
|
def set_require_ssp(self, val):
|
self._set_bool('require_ssp', val)
|
|
|
def set_build(self, val):
|
self._set_string('build', val)
|
|
|
def set_builds(self, val):
|
if type(val) == dict:
|
setattr(self, 'builds', val)
|
|
def set_max_result_size_kb(self, val):
|
self._set_int('max_result_size_KB', val)
|
|
def set_priority(self, val):
|
self._set_int('priority', val)
|
|
def set_fast(self, val):
|
self._set_bool('fast', val)
|
|
def set_attributes(self, val):
|
# Add subsystem:default if subsystem is not specified.
|
self._set_set('attributes', val)
|
if not any(a.startswith('subsystem') for a in self.attributes):
|
self.attributes.add('subsystem:default')
|
|
|
def _extract_const(expr):
|
assert(expr.__class__ == compiler.ast.Const)
|
assert(expr.value.__class__ in (str, int, float, unicode))
|
return str(expr.value).strip()
|
|
|
def _extract_dict(expr):
|
assert(expr.__class__ == compiler.ast.Dict)
|
assert(expr.items.__class__ == list)
|
cf_dict = {}
|
for key, value in expr.items:
|
try:
|
key = _extract_const(key)
|
val = _extract_expression(value)
|
except (AssertionError, ValueError):
|
pass
|
else:
|
cf_dict[key] = val
|
return cf_dict
|
|
|
def _extract_list(expr):
|
assert(expr.__class__ == compiler.ast.List)
|
list_values = []
|
for value in expr.nodes:
|
try:
|
list_values.append(_extract_expression(value))
|
except (AssertionError, ValueError):
|
pass
|
return list_values
|
|
|
def _extract_name(expr):
|
assert(expr.__class__ == compiler.ast.Name)
|
assert(expr.name in ('False', 'True', 'None'))
|
return str(expr.name)
|
|
|
def _extract_expression(expr):
|
if expr.__class__ == compiler.ast.Const:
|
return _extract_const(expr)
|
if expr.__class__ == compiler.ast.Name:
|
return _extract_name(expr)
|
if expr.__class__ == compiler.ast.Dict:
|
return _extract_dict(expr)
|
if expr.__class__ == compiler.ast.List:
|
return _extract_list(expr)
|
raise ValueError('Unknown rval %s' % expr)
|
|
|
def _extract_assignment(n):
|
assert(n.__class__ == compiler.ast.Assign)
|
assert(n.nodes.__class__ == list)
|
assert(len(n.nodes) == 1)
|
assert(n.nodes[0].__class__ == compiler.ast.AssName)
|
assert(n.nodes[0].flags.__class__ == str)
|
assert(n.nodes[0].name.__class__ == str)
|
|
val = _extract_expression(n.expr)
|
key = n.nodes[0].name.lower()
|
|
return (key, val)
|
|
|
def parse_control_string(control, raise_warnings=False, path=''):
|
"""Parse a control file from a string.
|
|
@param control: string containing the text of a control file.
|
@param raise_warnings: True iff ControlData should raise an error on
|
warnings about control file contents.
|
@param path: string path to the control file.
|
|
"""
|
try:
|
mod = compiler.parse(control)
|
except SyntaxError as e:
|
logging.error('Syntax error (%s) while parsing control string:', e)
|
lines = control.split('\n')
|
for n, l in enumerate(lines):
|
logging.error('Line %d: %s', n + 1, l)
|
raise ControlVariableException("Error parsing data because %s" % e)
|
return finish_parse(mod, path, raise_warnings)
|
|
|
def parse_control(path, raise_warnings=False):
|
try:
|
mod = compiler.parseFile(path)
|
except SyntaxError, e:
|
raise ControlVariableException("Error parsing %s because %s" %
|
(path, e))
|
return finish_parse(mod, path, raise_warnings)
|
|
|
def _try_extract_assignment(node, variables):
|
"""Try to extract assignment from the given node.
|
|
@param node: An Assign object.
|
@param variables: Dictionary to store the parsed assignments.
|
"""
|
try:
|
key, val = _extract_assignment(node)
|
variables[key] = val
|
except (AssertionError, ValueError):
|
pass
|
|
|
def finish_parse(mod, path, raise_warnings):
|
assert(mod.__class__ == compiler.ast.Module)
|
assert(mod.node.__class__ == compiler.ast.Stmt)
|
assert(mod.node.nodes.__class__ == list)
|
|
variables = {}
|
injection_variables = {}
|
for n in mod.node.nodes:
|
if (n.__class__ == compiler.ast.Function and
|
re.match('step\d+', n.name)):
|
vars_in_step = {}
|
for sub_node in n.code.nodes:
|
_try_extract_assignment(sub_node, vars_in_step)
|
if vars_in_step:
|
# Empty the vars collection so assignments from multiple steps
|
# won't be mixed.
|
variables.clear()
|
variables.update(vars_in_step)
|
else:
|
_try_extract_assignment(n, injection_variables)
|
|
variables.update(injection_variables)
|
return ControlData(variables, path, raise_warnings)
|
