| # 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, textwrap |
| |
| from autotest_lib.client.common_lib import enum |
| |
| 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:' |
| |
| 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.retries = 0 |
| 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() |
| |
| _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() |
| |
| |
| 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_retries(self, val): |
| self._set_int('retries', 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_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, e: |
| 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 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) |
| |
| vars = {} |
| for n in mod.node.nodes: |
| try: |
| key, val = _extract_assignment(n) |
| vars[key] = val |
| except (AssertionError, ValueError): |
| pass |
| return ControlData(vars, path, raise_warnings) |
