# -*- coding: utf-8 -*- "option types and option description for the configuration management" # Copyright (C) 2012 Team tiramisu (see README for all contributors) # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 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 General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # # The original `Config` design model is unproudly borrowed from # the rough gus of pypy: pypy: http://codespeak.net/svn/pypy/dist/pypy/config/ # the whole pypy projet is under MIT licence # ____________________________________________________________ from autolib import special_owners from basetype import HiddenBaseType, DisabledBaseType, ModeBaseType, modes from error import (ConfigError, ConflictConfigError, NotFoundError, RequiresError) available_actions = ['hide', 'show', 'enable', 'disable'] reverse_actions = {'hide': 'show', 'show': 'hide', 'disable':'enable', 'enable': 'disable'} # ____________________________________________________________ # OptionDescription authorized group_type values group_types = ['default', 'family', 'group', 'master'] # ____________________________________________________________ class Option(HiddenBaseType, DisabledBaseType, ModeBaseType): #reminder: an Option object is **not** a container for the value _frozen = False def __init__(self, name, doc, default=None, default_multi=None, requires=None, mandatory=False, multi=False, callback=None, callback_params=None, mode='normal'): self._name = name self.doc = doc self._requires = requires self._mandatory = mandatory self.multi = multi if not self.multi and default_multi is not None: raise ConfigError("a default_multi is set whereas multi is False" " in option: {0}".format(name)) self.default_multi = default_multi #if self.multi and default_multi is None: # _cfgimpl_warnings[name] = DefaultMultiWarning self.callback = callback if self.callback is None and callback_params is not None: raise ConfigError("params defined for a callback function but" " no callback defined yet for option {0}".format(name)) self.callback_params = callback_params if mode not in modes: raise ConfigError("mode {0} not available".format(mode)) self.mode = mode _default = default if default != None: if self.multi == True and type(default) != list: raise ConfigError("invalid default value {0} " "for option {1} : not list type".format(str(default), name)) if not self.validate(_default): raise ConfigError("invalid default value {0} " "for option {1}".format(str(default), name)) self.default = _default def validate(self, value): if self.multi == False: # None allows the reset of the value if value != None: return self._validate(value) else: if type(value) != list: raise ConfigError("invalid value {0} " "for option {1} which must be a list".format(value, self._name)) for val in value: if val != None: # None allows the reset of the value return self._validate(val) return True def getdefault(self): return self.default def getdoc(self): return self.doc def getcallback(self): return self.callback def getcallback_params(self): return self.callback_params def setowner(self, config, who): name = self._name if self._frozen: raise TypeError("trying to change a frozen option's owner: %s" % name) if who in special_owners: if self.callback == None: raise SpecialOwnersError("no callback specified for" "option {0}".format(name)) config._cfgimpl_value_owners[name] = who def setoption(self, config, value, who): name = self._name if self._frozen: raise TypeError('trying to change a frozen option object: %s' % name) # we want the possibility to reset everything if who == "default" and value is None: self.default = None return if not self.validate(value): raise ConfigError('invalid value %s for option %s' % (value, name)) if who == "default": # changes the default value (and therefore resets the previous value) if self._validate(value): self.default = value else: raise ConfigError("invalid value %s for option %s" % (value, name)) apply_requires(self, config) # FIXME put the validation for the multi somewhere else # # it is a multi **and** it has requires # if self.multi == True: # if type(value) != list: # raise TypeError("value {0} must be a list".format(value)) # if self._requires is not None: # for reqname in self._requires: # # FIXME : verify that the slaves are all multi # #option = getattr(config._cfgimpl_descr, reqname) # # if not option.multi == True: # # raise ConflictConfigError("an option with requires " # # "has to be a list type : {0}".format(name)) # if len(config._cfgimpl_values[reqname]) != len(value): # raise ConflictConfigError("an option with requires " # "has not the same length of the others " # "in the group : {0}".format(reqname)) config._cfgimpl_previous_values[name] = config._cfgimpl_values[name] config._cfgimpl_values[name] = value def getkey(self, value): return value def freeze(self): self._frozen = True return True def unfreeze(self): self._frozen = False # ____________________________________________________________ def is_multi(self): return self.multi def is_mandatory(self): return self._mandatory class ChoiceOption(Option): opt_type = 'string' def __init__(self, name, doc, values, default=None, requires=None, callback=None, callback_params=None, multi=False, mandatory=False): self.values = values super(ChoiceOption, self).__init__(name, doc, default=default, callback=callback, callback_params=callback_params, requires=requires, multi=multi, mandatory=mandatory) def setoption(self, config, value, who): name = self._name super(ChoiceOption, self).setoption(config, value, who) def _validate(self, value): return value is None or value in self.values class BoolOption(Option): opt_type = 'bool' # def __init__(self, name, doc, default=None, requires=None, # validator=None, multi=False, mandatory=False): # super(BoolOption, self).__init__(name, doc, default=default, # requires=requires, multi=multi, mandatory=mandatory) #self._validator = validator def _validate(self, value): return isinstance(value, bool) # FIXME config level validator # def setoption(self, config, value, who): # name = self._name # if value and self._validator is not None: # toplevel = config._cfgimpl_get_toplevel() # self._validator(toplevel) # super(BoolOption, self).setoption(config, value, who) class IntOption(Option): opt_type = 'int' def _validate(self, value): try: int(value) except TypeError: return False return True def setoption(self, config, value, who): try: super(IntOption, self).setoption(config, value, who) except TypeError, e: raise ConfigError(*e.args) class FloatOption(Option): opt_type = 'float' def _validate(self, value): try: float(value) except TypeError: return False return True def setoption(self, config, value, who): try: super(FloatOption, self).setoption(config, float(value), who) except TypeError, e: raise ConfigError(*e.args) class StrOption(Option): opt_type = 'string' def _validate(self, value): return isinstance(value, str) def setoption(self, config, value, who): try: super(StrOption, self).setoption(config, value, who) except TypeError, e: raise ConfigError(*e.args) class SymLinkOption(object): #(HiddenBaseType, DisabledBaseType): opt_type = 'symlink' def __init__(self, name, path): self._name = name self.path = path def setoption(self, config, value, who): try: setattr(config, self.path, value) # .setoption(self.path, value, who) except TypeError, e: raise ConfigError(*e.args) class IPOption(Option): opt_type = 'ip' def _validate(self, value): # by now the validation is nothing but a string, use IPy instead return isinstance(value, str) def setoption(self, config, value, who): try: super(IPOption, self).setoption(config, value, who) except TypeError, e: raise ConfigError(*e.args) class NetmaskOption(Option): opt_type = 'netmask' def _validate(self, value): # by now the validation is nothing but a string, use IPy instead return isinstance(value, str) def setoption(self, config, value, who): try: super(NetmaskOption, self).setoption(config, value, who) except TypeError, e: raise ConfigError(*e.args) class ArbitraryOption(Option): def __init__(self, name, doc, default=None, defaultfactory=None, requires=None, multi=False, mandatory=False): super(ArbitraryOption, self).__init__(name, doc, requires=requires, multi=multi, mandatory=mandatory) self.defaultfactory = defaultfactory if defaultfactory is not None: assert default is None def _validate(self, value): return True def getdefault(self): if self.defaultfactory is not None: return self.defaultfactory() return self.default class OptionDescription(HiddenBaseType, DisabledBaseType, ModeBaseType): group_type = 'default' def __init__(self, name, doc, children, requires=None): self._name = name self.doc = doc self._children = children self._requires = requires self._build() def getdoc(self): return self.doc def _build(self): for child in self._children: setattr(self, child._name, child) def add_child(self, child): "dynamically adds a configuration option" #Nothing is static. Even the Mona Lisa is falling apart. for ch in self._children: if isinstance(ch, Option): if child._name == ch._name: raise ConflictConfigError("existing option : {0}".format( child._name)) self._children.append(child) setattr(self, child._name, child) def update_child(self, child): "modification of an existing option" # XXX : corresponds to the `redefine`, is it usefull pass def getkey(self, config): return tuple([child.getkey(getattr(config, child._name)) for child in self._children]) def getpaths(self, include_groups=False, currpath=None): """returns a list of all paths in self, recursively currpath should not be provided (helps with recursion) """ if currpath is None: currpath = [] paths = [] for option in self._children: attr = option._name if attr.startswith('_cfgimpl'): continue value = getattr(self, attr) if isinstance(value, OptionDescription): if include_groups: paths.append('.'.join(currpath + [attr])) currpath.append(attr) paths += value.getpaths(include_groups=include_groups, currpath=currpath) currpath.pop() else: paths.append('.'.join(currpath + [attr])) return paths # ____________________________________________________________ def set_group_type(self, group_type): if group_type in group_types: self.group_type = group_type else: raise ConfigError('not allowed value for group_type : {0}'.format( group_type)) def get_group_type(self): return self.group_type # ____________________________________________________________ def hide(self): super(OptionDescription, self).hide() # FIXME : AND THE SUBCHILDREN ? for child in self._children: if isinstance(child, OptionDescription): child.hide() def show(self): # FIXME : AND THE SUBCHILDREN ?? super(OptionDescription, self).show() for child in self._children: if isinstance(child, OptionDescription): child.show() # ____________________________________________________________ def disable(self): super(OptionDescription, self).disable() # FIXME : AND THE SUBCHILDREN ? for child in self._children: if isinstance(child, OptionDescription): child.disable() def enable(self): # FIXME : AND THE SUBCHILDREN ? super(OptionDescription, self).enable() for child in self._children: if isinstance(child, OptionDescription): child.enable() # ____________________________________________________________ def apply_requires(opt, config): if hasattr(opt, '_requires'): if opt._requires is not None: # malformed requirements rootconfig = config._cfgimpl_get_toplevel() for req in opt._requires: if not type(req) == tuple and len(req) in (3, 4): raise RequiresError("malformed requirements for option:" " {0}".format(opt._name)) # all actions **must** be identical actions = [req[2] for req in opt._requires] action = actions[0] for act in actions: if act != action: raise RequiresError("malformed requirements for option:" " {0}".format(opt._name)) # filters the callbacks matches = False for req in opt._requires: if len(req) == 3: name, expected, action = req inverted = False if len(req) == 4: name, expected, action, inverted = req if inverted == 'inverted': inverted = True homeconfig, shortname = \ rootconfig._cfgimpl_get_home_by_path(name) # FIXME: doesn't work with 'auto' or 'fill' yet # (copy the code from the __getattr__ if shortname in homeconfig._cfgimpl_values: value = homeconfig._cfgimpl_values[shortname] if (not inverted and value == expected) or \ (inverted and value != expected): if action not in available_actions: raise RequiresError("malformed requirements" " for option: {0}".format(opt._name)) getattr(opt, action)() #.hide() or show() or... matches = True else: # option doesn't exist ! should not happen... raise NotFoundError("required option not found: " "{0}".format(name)) # no callback has been triggered, then just reverse the action if not matches: getattr(opt, reverse_actions[action])()