# -*- coding: utf-8 -*- # Copyright (C) 2014 Team tiramisu (see AUTHORS for all contributors) # # This program is free software: you can redistribute it and/or modify it # under the terms of the GNU Lesser General Public License as published by the # Free Software Foundation, either version 3 of the License, or (at your # option) any later version. # # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # details. # # You should have received a copy of the GNU Lesser General Public License # along with this program. If not, see . # # The original `Config` design model is unproudly borrowed from # the rough pypy's guys: http://codespeak.net/svn/pypy/dist/pypy/config/ # the whole pypy projet is under MIT licence # ____________________________________________________________ import re from types import FunctionType import warnings from ..i18n import _ from ..setting import log, undefined from ..autolib import carry_out_calculation from ..error import ConfigError, ValueWarning, PropertiesOptionError,\ ContextError from ..storage import get_storages_option StorageBase = get_storages_option('base') submulti = 2 name_regexp = re.compile(r'^[a-z][a-zA-Z\d\-_]*$') forbidden_names = frozenset(['iter_all', 'iter_group', 'find', 'find_first', 'make_dict', 'unwrap_from_path', 'read_only', 'read_write', 'getowner', 'set_contexts']) def valid_name(name): "an option's name is a str and does not start with 'impl' or 'cfgimpl'" if not isinstance(name, str): # pragma: optional cover return False if re.match(name_regexp, name) is not None and \ name not in forbidden_names and \ not name.startswith('impl_') and \ not name.startswith('cfgimpl_'): return True else: # pragma: optional cover return False def validate_callback(callback, callback_params, type_): if type(callback) != FunctionType: # pragma: optional cover raise ValueError(_('{0} must be a function').format(type_)) if callback_params is not None: if not isinstance(callback_params, dict): # pragma: optional cover raise ValueError(_('{0}_params must be a dict').format(type_)) for key, callbacks in callback_params.items(): if key != '' and len(callbacks) != 1: # pragma: optional cover raise ValueError(_("{0}_params with key {1} mustn't have " "length different to 1").format(type_, key)) if not isinstance(callbacks, tuple): # pragma: optional cover raise ValueError(_('{0}_params must be tuple for key "{1}"' ).format(type_, key)) for callbk in callbacks: if isinstance(callbk, tuple): if len(callbk) == 1: if callbk != (None,): # pragma: optional cover raise ValueError(_('{0}_params with length of ' 'tuple as 1 must only have ' 'None as first value')) elif len(callbk) != 2: # pragma: optional cover raise ValueError(_('{0}_params must only have 1 or 2 ' 'as length')) else: option, force_permissive = callbk if not isinstance(option, Option) and not \ isinstance(option, SymLinkOption): # pragma: optional cover raise ValueError(_('{0}_params must have an option' ' not a {0} for first argument' ).format(type_, type(option))) if force_permissive not in [True, False]: # pragma: optional cover raise ValueError(_('{0}_params must have a boolean' ' not a {0} for second argument' ).format(type_, type( force_permissive))) #____________________________________________________________ # class Base(StorageBase): __slots__ = tuple() def __init__(self, name, doc, default=None, default_multi=None, requires=None, multi=False, callback=None, callback_params=None, validator=None, validator_params=None, properties=None, warnings_only=False, extra=None, allow_empty_list=undefined): if not valid_name(name): # pragma: optional cover raise ValueError(_("invalid name: {0} for option").format(name)) if not multi and default_multi is not None: # pragma: optional cover raise ValueError(_("default_multi is set whereas multi is False" " in option: {0}").format(name)) if multi is True: is_multi = True _multi = 0 elif multi is False: is_multi = False _multi = 1 elif multi is submulti: is_multi = True _multi = submulti else: raise ValueError(_('invalid multi value')) if requires is not None: calc_properties, requires = validate_requires_arg(is_multi, requires, name) else: calc_properties = frozenset() requires = undefined if properties is None: properties = tuple() if not isinstance(properties, tuple): # pragma: optional cover raise TypeError(_('invalid properties type {0} for {1},' ' must be a tuple').format( type(properties), name)) if validator is not None: validate_callback(validator, validator_params, 'validator') self._set_validator(validator, validator_params) if calc_properties != frozenset([]) and properties is not tuple(): # pragma: optional cover set_forbidden_properties = calc_properties & set(properties) if set_forbidden_properties != frozenset(): raise ValueError('conflict: properties already set in ' 'requirement {0}'.format( list(set_forbidden_properties))) StorageBase.__init__(self, name, _multi, warnings_only, doc, extra, calc_properties, requires, properties, allow_empty_list) if multi is not False and default is None: default = [] self.impl_validate(default, is_multi=is_multi) self._set_default_values(default, default_multi, is_multi) ##callback is False in optiondescription if callback is not False: self.impl_set_callback(callback, callback_params, _init=True) self.commit() def impl_set_callback(self, callback, callback_params=None, _init=False): if callback is None and callback_params is not None: # pragma: optional cover raise ValueError(_("params defined for a callback function but " "no callback defined" " yet for option {0}").format( self.impl_getname())) if not _init and self.impl_get_callback()[0] is not None: raise ConfigError(_("a callback is already set for option {0}, " "cannot set another one's").format(self.impl_getname())) self._validate_callback(callback, callback_params) if callback is not None: validate_callback(callback, callback_params, 'callback') self._set_callback(callback, callback_params) def impl_is_optiondescription(self): return self.__class__.__name__ in ['OptionDescription', 'DynOptionDescription', 'SynDynOptionDescription'] def impl_is_dynoptiondescription(self): return self.__class__.__name__ in ['DynOptionDescription', 'SynDynOptionDescription'] class BaseOption(Base): """This abstract base class stands for attribute access in options that have to be set only once, it is of course done in the __setattr__ method """ __slots__ = tuple() # ____________________________________________________________ # serialize object def _impl_convert_requires(self, descr, load=False): """export of the requires during the serialization process :type descr: :class:`tiramisu.option.OptionDescription` :param load: `True` if we are at the init of the option description :type load: bool """ if not load and self.impl_getrequires() == []: self._state_requires = None elif load and self._state_requires is None: del(self._state_requires) else: if load: _requires = self._state_requires else: _requires = self.impl_getrequires() new_value = [] for requires in _requires: new_requires = [] for require in requires: if load: new_require = [descr.impl_get_opt_by_path(require[0])] else: new_require = [descr.impl_get_path_by_opt(require[0])] new_require.extend(require[1:]) new_requires.append(tuple(new_require)) new_value.append(tuple(new_requires)) if load: del(self._state_requires) if new_value != []: self._requires = new_value else: self._state_requires = new_value def _impl_convert_callback(self, descr, load=False): if self.__class__.__name__ == 'OptionDescription' or \ isinstance(self, SymLinkOption): return if not load and self.impl_get_callback() == (None, {}): self._state_callback = None self._state_callback_params = {} elif load and self._state_callback is None: del(self._state_callback) del(self._state_callback_params) else: if load: callback = self._state_callback callback_params = self._state_callback_params else: callback, callback_params = self.impl_get_callback() self._state_callback_params = {} cllbck_prms = {} for key, values in callback_params.items(): vls = [] for value in values: if isinstance(value, tuple) and value[0] is not None: if load: value = (descr.impl_get_opt_by_path(value[0]), value[1]) else: value = (descr.impl_get_path_by_opt(value[0]), value[1]) vls.append(value) cllbck_prms[key] = tuple(vls) if load: del(self._state_callback) del(self._state_callback_params) self._set_callback(callback, cllbck_prms) else: self._state_callback = callback self._state_callback_params = cllbck_prms # serialize def _impl_getstate(self, descr): """the under the hood stuff that need to be done before the serialization. :param descr: the parent :class:`tiramisu.option.OptionDescription` """ self._stated = True for func in dir(self): if func.startswith('_impl_convert_'): getattr(self, func)(descr) def __getstate__(self, stated=True): """special method to enable the serialization with pickle Usualy, a `__getstate__` method does'nt need any parameter, but somme under the hood stuff need to be done before this action :parameter stated: if stated is `True`, the serialization protocol can be performed, not ready yet otherwise :parameter type: bool """ try: self._stated except AttributeError: # pragma: optional cover raise SystemError(_('cannot serialize Option, ' 'only in OptionDescription')) if isinstance(self, SymLinkOption): slots = frozenset(['_name', '_state_opt', '_stated']) else: slots = self._impl_getattributes() slots -= frozenset(['_cache_paths', '_cache_consistencies', '__weakref__']) states = {} for slot in slots: # remove variable if save variable converted # in _state_xxxx variable if '_state' + slot not in slots: try: if slot.startswith('_state'): states[slot] = getattr(self, slot) # remove _state_xxx variable self.__delattr__(slot) else: states[slot] = getattr(self, slot) except AttributeError: pass if not stated: del(states['_stated']) return states # unserialize def _impl_setstate(self, descr): """the under the hood stuff that need to be done before the serialization. :type descr: :class:`tiramisu.option.OptionDescription` """ for func in dir(self): if func.startswith('_impl_convert_'): getattr(self, func)(descr, load=True) try: del(self._stated) except AttributeError: # pragma: optional cover pass def __setstate__(self, state): """special method that enables us to serialize (pickle) Usualy, a `__setstate__` method does'nt need any parameter, but somme under the hood stuff need to be done before this action :parameter state: a dict is passed to the loads, it is the attributes of the options object :type state: dict """ for key, value in state.items(): setattr(self, key, value) def __setattr__(self, name, value): """set once and only once some attributes in the option, like `_name`. `_name` cannot be changed one the option and pushed in the :class:`tiramisu.option.OptionDescription`. if the attribute `_readonly` is set to `True`, the option is "frozen" (which has noting to do with the high level "freeze" propertie or "read_only" property) """ if name not in ('_option', '_is_build_cache') and \ not isinstance(value, tuple) and \ not name.startswith('_state') and \ not name == '_sa_instance_state': is_readonly = False # never change _name dans _opt if name == '_name': try: if self.impl_getname() is not None: #so _name is already set is_readonly = True except (KeyError, AttributeError): pass elif name == '_opt': pass elif name != '_readonly': is_readonly = self.impl_is_readonly() if is_readonly: # pragma: optional cover raise AttributeError(_("'{0}' ({1}) object attribute '{2}' is" " read-only").format( self.__class__.__name__, self, #self.impl_getname(), name)) super(BaseOption, self).__setattr__(name, value) def impl_getpath(self, context): return context.cfgimpl_get_description().impl_get_path_by_opt(self) def impl_has_callback(self): "to know if a callback has been defined or not" return self.impl_get_callback()[0] is not None def _is_subdyn(self): return getattr(self, '_subdyn', None) is not None def impl_getproperties(self): return self._properties def _impl_valid_unicode(self, value): if not isinstance(value, unicode) and not isinstance(value, str): raise ValueError(_('invalid unicode or string')) class OnlyOption(BaseOption): __slots__ = tuple() class Option(OnlyOption): """ Abstract base class for configuration option's. Reminder: an Option object is **not** a container for the value. """ # __slots__ = ('_multi', '_validator', '_default_multi', '_default', # '_state_callback', '_callback', # '_consistencies', '_warnings_only', '_master_slaves', # '_state_consistencies', '__weakref__') __slots__ = tuple() _empty = '' def _launch_consistency(self, func, option, value, context, index, submulti_index, all_cons_opts, warnings_only, transitive): """Launch consistency now :param func: function name, this name should start with _cons_ :type func: `str` :param option: option that value is changing :type option: `tiramisu.option.Option` :param value: new value of this option :param context: Config's context, if None, check default value instead :type context: `tiramisu.config.Config` :param index: only for multi option, consistency should be launch for specified index :type index: `int` :param all_cons_opts: all options concerne by this consistency :type all_cons_opts: `list` of `tiramisu.option.Option` :param warnings_only: specific raise error for warning :type warnings_only: `boolean` :param transitive: propertyerror is transitive :type transitive: `boolean` """ if context is not undefined: descr = context.cfgimpl_get_description() all_cons_vals = [] for opt in all_cons_opts: try: #get value if (isinstance(opt, DynSymLinkOption) and option._dyn == opt._dyn) or \ option == opt: opt_value = value else: #if context, calculate value, otherwise get default value if context is not undefined: if isinstance(opt, DynSymLinkOption): path = opt.impl_getpath(context) else: path = descr.impl_get_path_by_opt(opt) opt_value = context.getattr(path, validate=False, force_permissive=True) else: opt_value = opt.impl_getdefault() #append value if not self.impl_is_multi() or (isinstance(opt, DynSymLinkOption) and option._dyn == opt._dyn) or \ option == opt: all_cons_vals.append(opt_value) elif self.impl_is_submulti(): try: all_cons_vals.append(opt_value[index][submulti_index]) except IndexError, err: log.debug('indexerror in submulti opt in _launch_consistency: {0}'.format(err)) #value is not already set, could be higher index #so return if no value return else: try: all_cons_vals.append(opt_value[index]) except IndexError, err: #value is not already set, could be higher index #so return if no value and not default_value log.debug('indexerror in _launch_consistency: {0}'.format(err)) return except PropertiesOptionError as err: log.debug('propertyerror in _launch_consistency: {0}'.format(err)) if transitive: raise err else: return getattr(self, func)(all_cons_opts, all_cons_vals, warnings_only) def impl_validate(self, value, context=undefined, validate=True, force_index=None, force_submulti_index=None, current_opt=undefined, is_multi=None): """ :param value: the option's value :param context: Config's context :type context: :class:`tiramisu.config.Config` :param validate: if true enables ``self._validator`` validation :type validate: boolean :param force_index: if multi, value has to be a list not if force_index is not None :type force_index: integer :param force_submulti_index: if submulti, value has to be a list not if force_submulti_index is not None :type force_submulti_index: integer """ if not validate: return if current_opt is undefined: current_opt = self def calculation_validator(val): validator, validator_params = self.impl_get_validator() if validator is not None: if validator_params != {}: validator_params_ = {} for val_param, values in validator_params.items(): validator_params_[val_param] = values #inject value in calculation if '' in validator_params_: lst = list(validator_params_['']) lst.insert(0, val) validator_params_[''] = tuple(lst) else: validator_params_[''] = (val,) else: validator_params_ = {'': (val,)} # Raise ValueError if not valid try: carry_out_calculation(current_opt, context=context, callback=validator, callback_params=validator_params_) except ContextError: pass def do_validation(_value, _index, submulti_index): if _value is None: return # option validation try: self._validate(_value, context, current_opt) except ValueError as err: # pragma: optional cover log.debug('do_validation: value: {0}, index: {1}, ' 'submulti_index: {2}'.format(_value, _index, submulti_index), exc_info=True) raise ValueError(_('invalid value for option {0}: {1}' '').format(self.impl_getname(), err)) error = None warning = None try: # valid with self._validator calculation_validator(_value) self._second_level_validation(_value, self._is_warnings_only()) except ValueError as error: log.debug(_('do_validation for {0}: error in value').format( self.impl_getname()), exc_info=True) if self._is_warnings_only(): warning = error error = None if error is None and warning is None: try: # if context launch consistency validation #if context is not undefined: self._valid_consistency(current_opt, _value, context, _index, submulti_index) except ValueError as error: log.debug(_('do_validation for {0}: error in consistency').format( self.impl_getname()), exc_info=True) pass except ValueWarning as warning: log.debug(_('do_validation for {0}: warning in consistency').format( self.impl_getname()), exc_info=True) pass if warning: msg = _("warning on the value of the option {0}: {1}").format( self.impl_getname(), warning) if context is undefined or 'warnings' in \ context.cfgimpl_get_settings(): warnings.warn_explicit(ValueWarning(msg, self), ValueWarning, self.__class__.__name__, 0) elif error: raise ValueError(_("invalid value for option {0}: {1}").format( self.impl_getname(), error)) # generic calculation #if context is not undefined: # descr = context.cfgimpl_get_description() if is_multi is None: is_multi = self.impl_is_multi() if not is_multi: do_validation(value, None, None) elif force_index is not None: if self.impl_is_submulti() and force_submulti_index is None: if not isinstance(value, list): # pragma: optional cover raise ValueError(_("invalid value {0} for option {1} which" " must be a list").format( value, self.impl_getname())) for idx, val in enumerate(value): do_validation(val, force_index, idx) else: do_validation(value, force_index, force_submulti_index) else: if not isinstance(value, list): # pragma: optional cover raise ValueError(_("invalid value {0} for option {1} which " "must be a list").format(value, self.impl_getname())) for idx, val in enumerate(value): if self.impl_is_submulti() and force_submulti_index is None: if not isinstance(val, list): # pragma: optional cover raise ValueError(_("invalid value {0} for option {1} " "which must be a list of list" "").format(value, self.impl_getname())) for slave_idx, slave_val in enumerate(val): do_validation(slave_val, idx, slave_idx) else: do_validation(val, idx, force_submulti_index) def impl_is_master_slaves(self, type_='both'): """FIXME """ try: self._master_slaves if type_ in ('both', 'master') and \ self._master_slaves.is_master(self): return True if type_ in ('both', 'slave') and \ not self._master_slaves.is_master(self): return True except: pass return False def impl_get_master_slaves(self): return self._master_slaves def impl_getdoc(self): "accesses the Option's doc" return self.impl_get_information('doc') def _valid_consistencies(self, other_opts): if self._is_subdyn(): dynod = self._impl_getsubdyn() else: dynod = None for opt in other_opts: if not isinstance(opt, Option): # pragma: optional cover raise ConfigError(_('consistency must be set with an option')) if opt._is_subdyn(): if dynod is None: raise ConfigError(_('almost one option in consistency is ' 'in a dynoptiondescription but not all')) if dynod != opt._impl_getsubdyn(): raise ConfigError(_('option in consistency must be in same' ' dynoptiondescription')) dynod = opt._impl_getsubdyn() elif dynod is not None: raise ConfigError(_('almost one option in consistency is in a ' 'dynoptiondescription but not all')) if self is opt: # pragma: optional cover raise ConfigError(_('cannot add consistency with itself')) if self.impl_is_multi() != opt.impl_is_multi(): # pragma: optional cover raise ConfigError(_('every options in consistency must be ' 'multi or none')) def impl_add_consistency(self, func, *other_opts, **params): """Add consistency means that value will be validate with other_opts option's values. :param func: function's name :type func: `str` :param other_opts: options used to validate value :type other_opts: `list` of `tiramisu.option.Option` :param params: extra params (warnings_only and transitive are allowed) """ if self.impl_is_readonly(): # pragma: optional cover raise AttributeError(_("'{0}' ({1}) cannot add consistency, option is" " read-only").format( self.__class__.__name__, self.impl_getname())) self._valid_consistencies(other_opts) func = '_cons_{0}'.format(func) if func not in dir(self): raise ConfigError(_('consistency {0} not available for this option').format(func)) all_cons_opts = tuple([self] + list(other_opts)) unknown_params = set(params.keys()) - set(['warnings_only', 'transitive']) if unknown_params != set(): raise ValueError(_('unknow parameter {0} in consistency').format(unknown_params)) self._add_consistency(func, all_cons_opts, params) #validate default value when add consistency try: self.impl_validate(self.impl_getdefault()) except ValueError, err: self._del_consistency() raise err def _valid_consistency(self, option, value, context, index, submulti_idx): if context is not undefined: descr = context.cfgimpl_get_description() if descr._cache_consistencies is None: return True #consistencies is something like [('_cons_not_equal', (opt1, opt2))] if isinstance(option, DynSymLinkOption): consistencies = descr._cache_consistencies.get(option._impl_getopt()) else: consistencies = descr._cache_consistencies.get(option) else: consistencies = option._get_consistencies() if consistencies is not None: for func, all_cons_opts, params in consistencies: warnings_only = params.get('warnings_only', False) transitive = params.get('transitive', True) #all_cons_opts[0] is the option where func is set if isinstance(option, DynSymLinkOption): subpath = '.'.join(option._dyn.split('.')[:-1]) namelen = len(option._impl_getopt().impl_getname()) suffix = option.impl_getname()[namelen:] opts = [] for opt in all_cons_opts: name = opt.impl_getname() + suffix path = subpath + '.' + name opts.append(opt._impl_to_dyn(name, path)) else: opts = all_cons_opts try: opts[0]._launch_consistency(func, option, value, context, index, submulti_idx, opts, warnings_only, transitive) except ValueError as err: if warnings_only: raise ValueWarning(err.message, option) else: raise err def _cons_not_equal(self, opts, vals, warnings_only): for idx_inf, val_inf in enumerate(vals): for idx_sup, val_sup in enumerate(vals[idx_inf + 1:]): if val_inf == val_sup is not None: if warnings_only: msg = _("same value for {0} and {1}, should be different") else: msg = _("same value for {0} and {1}, must be different") log.debug('_cons_not_equal: {0} and {1} are not different'.format(val_inf, val_sup)) raise ValueError(msg.format(opts[idx_inf].impl_getname(), opts[idx_inf + idx_sup + 1].impl_getname())) # serialize/unserialize def _impl_convert_consistencies(self, descr, load=False): """during serialization process, many things have to be done. one of them is the localisation of the options. The paths are set once for all. :type descr: :class:`tiramisu.option.OptionDescription` :param load: `True` if we are at the init of the option description :type load: bool """ if not load and self._get_consistencies() == (): self._state_consistencies = None elif load and self._state_consistencies is None: del(self._state_consistencies) else: if load: consistencies = self._state_consistencies else: consistencies = self._get_consistencies() new_value = [] for consistency in consistencies: values = [] for obj in consistency[1]: if load: values.append(descr.impl_get_opt_by_path(obj)) else: values.append(descr.impl_get_path_by_opt(obj)) new_value.append((consistency[0], tuple(values), consistency[2])) if load: del(self._state_consistencies) for new_val in new_value: self._add_consistency(new_val[0], new_val[1], new_val[2]) else: self._state_consistencies = new_value def _second_level_validation(self, value, warnings_only): pass def _impl_to_dyn(self, name, path): return DynSymLinkOption(name, self, dyn=path) def _validate_callback(self, callback, callback_params): """callback_params: * None * {'': ((option, permissive),), 'ip': ((None,), (option, permissive)) """ if callback is None: return default_multi = self.impl_getdefault_multi() is_multi = self.impl_is_multi() default = self.impl_getdefault() if (not is_multi and (default is not None or default_multi is not None)) or \ (is_multi and (default != [] or default_multi is not None)): # pragma: optional cover raise ValueError(_("default value not allowed if option: {0} " "is calculated").format(self.impl_getname())) def validate_requires_arg(multi, requires, name): """check malformed requirements and tranform dict to internal tuple :param requires: have a look at the :meth:`tiramisu.setting.Settings.apply_requires` method to know more about the description of the requires dictionary """ ret_requires = {} config_action = {} # start parsing all requires given by user (has dict) # transforme it to a tuple for require in requires: if not isinstance(require, dict): # pragma: optional cover raise ValueError(_("malformed requirements type for option:" " {0}, must be a dict").format(name)) valid_keys = ('option', 'expected', 'action', 'inverse', 'transitive', 'same_action') unknown_keys = frozenset(require.keys()) - frozenset(valid_keys) if unknown_keys != frozenset(): # pragma: optional cover raise ValueError(_('malformed requirements for option: {0}' ' unknown keys {1}, must only ' '{2}').format(name, unknown_keys, valid_keys)) # prepare all attributes try: option = require['option'] expected = require['expected'] action = require['action'] except KeyError: # pragma: optional cover raise ValueError(_("malformed requirements for option: {0}" " require must have option, expected and" " action keys").format(name)) if action == 'force_store_value': # pragma: optional cover raise ValueError(_("malformed requirements for option: {0}" " action cannot be force_store_value" ).format(name)) inverse = require.get('inverse', False) if inverse not in [True, False]: # pragma: optional cover raise ValueError(_('malformed requirements for option: {0}' ' inverse must be boolean')) transitive = require.get('transitive', True) if transitive not in [True, False]: # pragma: optional cover raise ValueError(_('malformed requirements for option: {0}' ' transitive must be boolean')) same_action = require.get('same_action', True) if same_action not in [True, False]: # pragma: optional cover raise ValueError(_('malformed requirements for option: {0}' ' same_action must be boolean')) if not isinstance(option, Option): # pragma: optional cover raise ValueError(_('malformed requirements ' 'must be an option in option {0}').format(name)) if not multi and option.impl_is_multi(): raise ValueError(_('malformed requirements ' 'multi option must not set ' 'as requires of non multi option {0}').format(name)) if expected is not None: try: option._validate(expected) except ValueError as err: # pragma: optional cover raise ValueError(_('malformed requirements second argument ' 'must be valid for option {0}' ': {1}').format(name, err)) if action in config_action: if inverse != config_action[action]: # pragma: optional cover raise ValueError(_("inconsistency in action types" " for option: {0}" " action: {1}").format(name, action)) else: config_action[action] = inverse if action not in ret_requires: ret_requires[action] = {} if option not in ret_requires[action]: ret_requires[action][option] = (option, [expected], action, inverse, transitive, same_action) else: ret_requires[action][option][1].append(expected) # transform dict to tuple ret = [] for opt_requires in ret_requires.values(): ret_action = [] for require in opt_requires.values(): ret_action.append((require[0], tuple(require[1]), require[2], require[3], require[4], require[5])) ret.append(tuple(ret_action)) return frozenset(config_action.keys()), tuple(ret) class SymLinkOption(OnlyOption): # __slots__ = ('_opt', '_state_opt') def __init__(self, name, opt): if not isinstance(opt, Option): # pragma: optional cover raise ValueError(_('malformed symlinkoption ' 'must be an option ' 'for symlink {0}').format(name)) super(Base, self).__init__(name, undefined, undefined, undefined, undefined, undefined, undefined, undefined, undefined, opt) self.commit() def __getattr__(self, name, context=undefined): if name in ('_opt', '_readonly', 'impl_getpath', '_name', '_state_opt', '_impl_setopt'): return object.__getattr__(self, name) else: return getattr(self._impl_getopt(), name) def _impl_getstate(self, descr): self._stated = True self._state_opt = descr.impl_get_path_by_opt(self._impl_getopt()) def _impl_setstate(self, descr): self._impl_setopt(descr.impl_get_opt_by_path(self._state_opt)) del(self._state_opt) try: del(self._stated) except AttributeError: # pragma: optional cover pass self._set_readonly(True) def impl_get_information(self, key, default=undefined): return self._impl_getopt().impl_get_information(key, default) def impl_is_readonly(self): return True def impl_getproperties(self): return self._impl_getopt()._properties def impl_get_callback(self): return self._impl_getopt().impl_get_callback() def impl_has_callback(self): "to know if a callback has been defined or not" return self._impl_getopt().impl_has_callback() def impl_is_multi(self): return self._impl_getopt().impl_is_multi() def _is_subdyn(self): try: return self._impl_getopt()._subdyn is not None except AttributeError: return False class DynSymLinkOption(object): __slots__ = ('_dyn', '_opt', '_name') def __init__(self, name, opt, dyn): self._name = name self._dyn = dyn self._opt = opt def __getattr__(self, name, context=undefined): if name in ('_opt', '_readonly', 'impl_getpath', '_name', '_state_opt'): return object.__getattr__(self, name) else: return getattr(self._impl_getopt(), name) def impl_getname(self): return self._name def _impl_getopt(self): return self._opt def impl_getsuffix(self): return self._dyn.split('.')[-1][len(self._impl_getopt().impl_getname()):] def impl_getpath(self, context): path = self._impl_getopt().impl_getpath(context) base_path = '.'.join(path.split('.')[:-2]) if self.impl_is_master_slaves() and base_path is not '': base_path = base_path + self.impl_getsuffix() if base_path == '': return self._dyn else: return base_path + '.' + self._dyn def impl_validate(self, value, context=undefined, validate=True, force_index=None, force_submulti_index=None, is_multi=None): return self._impl_getopt().impl_validate(value, context, validate, force_index, force_submulti_index, current_opt=self, is_multi=is_multi)