Source code for pcvs.helpers.criterion

import base64
import itertools
import math
import os

from pcvs import io
from pcvs.helpers.exceptions import CommonException
from pcvs.helpers.system import MetaConfig
from pcvs.plugins import Plugin

[docs]class Combination: """A combination maps the actual concretization from multiple criterion. For a given set of criterion, a Combination carries, for each kind, its associated value in order to generate the appropriate test """ def __init__(self, crit_desc, dict_comb): """Build a combination from two components: - the actual combination dict - the dict of criterions :param crit_desc: dict of criterions (=their full description) represented in the current combination. :type crit_desc: dict :param dict_comb: actual combination dict (k=criterion name, v=actual value) :type dict_comb: dict """ self._criterions = crit_desc self._combination = dict_comb
[docs] def get(self, k, dflt=None): """Retrieve the actual value for a given combination element :param k: value to retrieve :type k: str :param dflt: default value if k is not a valid key :type: object """ if k not in self._combination: return dflt return self._combination[k]
[docs] def items(self): """Get the combination dict. :return: the whole combination dict. :rtype: dict """ return self._combination.items()
[docs] def translate_to_str(self): """Translate the actual combination in a pretty-format string. This is mainly used to generate actual test names """ c = self._criterions string = list() # each combination is built following: 'defined-prefix+value' for n in sorted(self._combination.keys()): subtitle = c[n].subtitle if subtitle is None: subtitle = "{}".format(n) string.append(subtitle + str(self._combination[n]).replace(" ", "-")) return "_".join(string)
[docs] def translate_to_command(self): """Translate the actual combination is tuple of three elements, based on the representation of each criterion in the test semantic. It builds tokens to provide to properly build the test command. It can either be: 1. an environment variable to export before the test to run (gathering system-scope and program-scope elements) 2. a runtime argument 3. a program-level argument (through custom-made iterators) """ args = [] envs = [] params = [] # for each elt, where k is the criterion name, v is the actual value for k_elt, v_elt in self._combination.items(): c = self._criterions[k_elt] # concretize_value() gathers both criterion label & value according # to specs (before, after, aliasing...) value = c.concretize_value(v_elt) if c.is_env(): envs.append(value) elif c.is_local(): params.append(value) else: args.append(value) return (envs, args, params)
[docs] def translate_to_dict(self): """Translate the combination into a dictionary. :return: configuration in the shape of a python dict :rtype: dict """ return self._combination
[docs]class Serie: """A serie ties a test expression (TEDescriptor) to the possible values which can be taken for each criterion to build test sets. A serie can be seen as the Combination generator for a given TEDescriptor """
[docs] @classmethod def register_sys_criterion(cls, system_criterion): """copy/inherit the system-defined criterion (shortcut to global config) """ cls.sys_iterators = system_criterion
def __init__(self, dict_of_criterion): """Build a serie, by extracting the list of values. Note that here, the dict also contains program-based criterions :param dict_of_criterion: values to build the serie with :type dict_of_criterion: dict""" self._values = list() self._keys = list() # this has to be saved, need to be forwarded to each combination self._dict = dict_of_criterion for name, node in dict_of_criterion.items(): assert (isinstance(node, Criterion)) assert (name == self._values.append(node.values) self._keys.append(
[docs] def generate(self): """Generator to build each combination""" for combination in itertools.product(*self._values): d = {self._keys[i]: val for i, val in enumerate(combination)} if not valid_combination(d): continue yield Combination( self._dict, d )
[docs]class Criterion: """A Criterion is the representation of a component each program (i.e. test binary) should be run against. A criterion comes with a range of possible values, each leading to a different test""" def __init__(self, name, description, local=False, numeric=False): """Initialize a criterion from its YAML/dict description :param name: name of the criterion :type name: str :param description: description of the criterion :type description: str :param local: True if the criterion is local, default to False :type local: bool :param numeric: True if the criterion is numeric, default to False :type: numeric: bool""" self._name = name if description is None: self._values = None return self._numeric = description.get('numeric', numeric) is True self._prefix = description.get('option', '') self._after = description.get('position', 'after') == 'after' self._alias = description.get('aliases', {}) self._is_env = description.get('type', 'argument') == 'environment' # this should be only set by per-TE criterion definition self._local = local self._str = description.get('subtitle', None) self._values = description.get('values', []) self._expanded = False #Sanity check self.sanitize_values()
[docs] def sanitize_values(self): """ Check for any inconsistent values in the current Criterion. Currently, only scalar items or dict (=> sequence) are allowed. Will raise an exeption in case of inconsistency (Maybe this should be managed in another way through the error handling) """ if self.is_discarded(): return if not isinstance(self._values, list): self._values = [self._values] for v in self._values: if isinstance(v, list): raise CommonException.UnclassifiableError( reason="list elements should be scalar OR dict", dbg_info={"element": v} ) if isinstance(v, dict): for key in v.keys(): assert key in ['op', 'of', 'from', 'to']
# only allow overriding values (for now)
[docs] def override(self, desc): """Replace the value of the criterion using a descriptor containing the said value :param desc: descriptor supposedly containing a ``value``entry :type desc: dict """ if 'values' in desc: self._values = desc['values'] self._expanded = False self.sanitize_values()
[docs] def intersect(self, other): """Update the calling Criterion with the interesection of the current range of possible values with the one given as a parameters. This is used to refine overriden per-TE criterion according to system-wide's""" assert (isinstance(other, Criterion)) assert (self._name == other._name) # None is special value meaning, discard this criterion because # irrelevant if self._values is None or other._values is None: self._values = None else: self._values = set(self._values).intersection(other._values)
[docs] def is_empty(self): """Is the current set of values empty May lead to errors, as it may indicates no common values has been found between user and system specifications""" return self._values is not None and len(self._values) == 0
[docs] def is_discarded(self): """Should this criterion be ignored from the current TE generaiton ?""" return self._values is None
[docs] def is_local(self): """Is the criterion local ? (program-scoped)""" return self._local
[docs] def is_env(self): """Is this criterion targeting a component used as an env var ?""" return self._is_env
@staticmethod def __convert_str_to_int(str_elt): """Convert a sequence (as a string) into a valid range of values. This is used to build criterion numeric-only possible values""" # TODO: write sequence conversion for numeric values return 0 @property def values(self): """Get the ``value`` attribute of this criterion. :return: values of this criterion :rtype: list """ return self._values def __len__(self): """Return the number of values this criterion holds. :return: the value list count :rtype: int """ return len(self._values) @property def name(self): """Get the ``name`` attribute of this criterion. :return: name of this criterion :rtype: str """ return self._name @property def subtitle(self): """Get the ``subtitle`` attribute of this criterion. :return: subtitle of this criterion :rtype: str """ return self._str @property def numeric(self): """Get the ``numeric`` attribute of this criterion. :return: numeric of this criterion :rtype: str """ return self._numeric
[docs] def concretize_value(self, val=''): """Return the exact string mapping this criterion, according to the specification. (is it aliased ? should the option be put before/after the value?...) :param val: value to add with prefix :type val: str :return: values with aliases replaced :rtype: str""" # replace value with alias (if defined) val = str(self.aliased_value(val)) # put value before of after the defined prefix elt = self._prefix + val if self._after else val + self._prefix # return the elt. up to the caller to determine # if this should be added as an arg or an env # ==> is_env() return elt
[docs] def aliased_value(self, val): """Check if the given value has an alias for the current criterion. An alias is the value replacement to use instead of the one defined by test configuration. This allows to split test logic from runtime semantics. For instance, TEs manipulate 'ib' as a value to depict the 'infiniband' network layer. But once the test has to be built, the term will change depending on the runtime carrying it, the value may be different from a runtime to another :param val: string with aliases to be replaced""" return self._alias[val] if val in self._alias else val
@staticmethod def __convert_sequence_to_list(node, s=-1, e=-1): """converts a sequence (as a string) to a valid list of values :param dic: dictionary to take the values from :type dic: dict :param s: start (can be overridden by ``from`` in ``dic``), defaults to -1 :type s: int, optional :param e: end (can be overridden by ``to`` in ``dic``), defaults to -1 :type e: int, optional :return: list of values :rtype: list """ values = [] # these must be integers def _convert_sequence_item_to_int(val): """helper to convert a string-formated number to a valid repr. :param val: the string-based number to convert :type val: str :raises CommonException.BadTokenError: val is not a number :return: the number :rtype: int() or float() """ if not isinstance(val, int) or not isinstance(val, float): try: n = float(val) if n.is_integer(): return int(n) else: return n except ValueError: raise CommonException.BadTokenError(val) else: return val start = _convert_sequence_item_to_int(node.get('from', s)) end = _convert_sequence_item_to_int(node.get('to', e)) of = _convert_sequence_item_to_int(node.get('of', 1)) op = node.get('op', 'seq').lower() if op in ['seq', 'arithmetic', 'ari']: values = range(start, end+1, of) elif op in ['mul', 'geometric', 'geo']: if start == 0: values.append(0) elif of in [-1, 0, 1]: values.append(start ** of) else: cur = start while cur <= end: values.append(cur) cur *= of elif op in ['pow', 'powerof']: if of == 0: values.append() start = math.ceil(start**(1/of)) end = math.floor(end**(1/of)) for i in range(start, end+1): values.append(i**of) else: io.console.warn("failure in Criterion sequence!") return values @property def expanded(self): return self._expanded @property def min_value(self): assert(self.expanded) return min(self._values) @property def max_value(self): assert(self.expanded) return max(self._values)
[docs] def expand_values(self, reference=None): """Browse values for the current criterion and make it ready to generate combinations""" values = [] start = 0 end = 100 if self.expanded: return if reference: assert isinstance(reference, Criterion) if not reference.expanded: reference.expand_values() start = reference.min_value end = reference.max_value io.console.debug("Expanding {}: {}".format(, self._values)) if self._numeric is True: for v in self._values: if isinstance(v, dict): values += self.__convert_sequence_to_list(v, s=start, e=end) elif isinstance(v, (int, float, str)): values.append(v) else: raise TypeError("Only accept int or sequence (as string)" " as values for numeric iterators") else: values = self._values # now ensure values are unique self._values = set(values) self._expanded = True io.console.debug("EXPANDED {}: {}".format(, self._values))
# TODO: handle criterion dependency (ex: n_mpi: ['n_node * 2'])
[docs]def initialize_from_system(): """Initialise system-wide criterions TODO: Move this function elsewhere.""" # sanity checks if not MetaConfig.root.criterion: MetaConfig.root.set_internal('crit_obj', {}) else: # raw YAML objects runtime_iterators = MetaConfig.root.runtime.criterions criterion_iterators = MetaConfig.root.criterion it_to_remove = [] # if a criterion defined in criterion.yaml but # not declared as part of a runtime, the criterion # should be silently discarded # here is the purpose for it in criterion_iterators.keys(): if it not in runtime_iterators: io.console.warn("Undeclared criterion " "as part of runtime: '{}' ".format(it)) elif criterion_iterators[it]['values'] is None: io.console.debug('No combination found for {},' ' removing from schedule'.format(it)) else: continue"Removing '{}'".format(it)) it_to_remove.append(it) # register the new dict {criterion_name: Criterion object} # the criterion object gathers both information from runtime & criterion MetaConfig.root.set_internal('crit_obj', {k: Criterion( k, {**runtime_iterators[k], **criterion_iterators[k]}) for k in criterion_iterators.keys() if k not in it_to_remove}) # convert any sequence into valid range of integers for # numeric criterions comb_cnt = 1 for criterion in MetaConfig.root.get_internal('crit_obj').values(): criterion.expand_values() comb_cnt *= len(criterion) MetaConfig.root.set_internal("comb_cnt", comb_cnt)
first = True
[docs]def valid_combination(dic): """Check if dict is a valid criterion combination . :param dic: dict to check :type dic: dict :return: True if dic is a valid combination :rtype: bool """ global first rt = MetaConfig.root.runtime val = MetaConfig.root.validation pCollection = MetaConfig.root.get_internal('pColl') if first and rt.plugin: first = not first rt.pluginfile = os.path.join(val.buildcache, "") with open(rt.pluginfile, 'w') as fh: fh.write(base64.b64decode(rt.plugin).decode('utf-8')) pCollection.register_plugin_by_file(rt.pluginfile, activate=True) ret = pCollection.invoke_plugins(Plugin.Step.TEST_EVAL, config=MetaConfig.root, combination=dic) # by default, no plugin = always true if ret is None: ret = True return ret