483 lines
18 KiB
Python
483 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
"pretty small and local configuration management tool"
|
|
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']
|
|
# Option and OptionDescription modes
|
|
modes = ['normal', 'expert']
|
|
# ____________________________________________________________
|
|
# interfaces
|
|
class HiddenBaseType(object):
|
|
hidden = False
|
|
def hide(self):
|
|
self.hidden = True
|
|
def show(self):
|
|
self.hidden = False
|
|
def _is_hidden(self):
|
|
# dangerous method: how an Option can determine its status by itself ?
|
|
return self.hidden
|
|
|
|
class DisabledBaseType(object):
|
|
disabled = False
|
|
def disable(self):
|
|
self.disabled = True
|
|
def enable(self):
|
|
self.disabled = False
|
|
def _is_disabled(self):
|
|
return self.disabled
|
|
|
|
class ModeBaseType(object):
|
|
mode = 'normal'
|
|
def get_mode(self):
|
|
return self.mode
|
|
def set_mode(self, mode):
|
|
if mode not in modes:
|
|
raise TypeError("Unknown mode: {0}".format(mode))
|
|
self.mode = mode
|
|
# ____________________________________________________________
|
|
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, requires=None,
|
|
mandatory=False, multi=False, callback=None, mode='normal'):
|
|
self._name = name
|
|
self.doc = doc
|
|
self._requires = requires
|
|
self._mandatory = mandatory
|
|
self.multi = multi
|
|
self.callback = callback
|
|
if mode not in modes:
|
|
raise ConfigError("mode {0} not available".format(mode))
|
|
self.mode = mode
|
|
if default != None:
|
|
if not self.validate(default):
|
|
raise ConfigError("invalid default value {0} "
|
|
"for option {1}".format(default, name))
|
|
self.default = default
|
|
|
|
def validate(self, value):
|
|
raise NotImplementedError('abstract base class')
|
|
|
|
def getdefault(self):
|
|
return self.default
|
|
|
|
def getdoc(self):
|
|
return self.doc
|
|
|
|
def getcallback(self):
|
|
return self.callback
|
|
|
|
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 ['auto', 'fill']: # XXX special_owners to be imported from config
|
|
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)
|
|
self.default = value
|
|
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,
|
|
multi=False, mandatory=False):
|
|
self.values = values
|
|
super(ChoiceOption, self).__init__(name, doc, default=default,
|
|
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):
|
|
if self.multi == False:
|
|
return value is None or value in self.values
|
|
else:
|
|
for val in value:
|
|
if not (val is None or val in self.values):
|
|
return False
|
|
return True
|
|
|
|
class BoolOption(Option):
|
|
opt_type = 'bool'
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(BoolOption, self).__init__(*args, **kwargs)
|
|
# 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):
|
|
if self.multi == False:
|
|
return isinstance(value, bool)
|
|
else:
|
|
try:
|
|
for val in value:
|
|
if not isinstance(val, bool):
|
|
return False
|
|
except Exception:
|
|
return False
|
|
return True
|
|
# 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 __init__(self, *args, **kwargs):
|
|
super(IntOption, self).__init__(*args, **kwargs)
|
|
|
|
def validate(self, value):
|
|
if self.multi == False:
|
|
try:
|
|
int(value)
|
|
except TypeError:
|
|
return False
|
|
return True
|
|
else:
|
|
for val in value:
|
|
try:
|
|
int(val)
|
|
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 __init__(self, *args, **kwargs):
|
|
super(FloatOption, self).__init__(*args, **kwargs)
|
|
|
|
def validate(self, value):
|
|
if self.multi == False:
|
|
try:
|
|
float(value)
|
|
except TypeError:
|
|
return False
|
|
return True
|
|
else:
|
|
for val in value:
|
|
try:
|
|
float(val)
|
|
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 __init__(self, *args, **kwargs):
|
|
super(StrOption, self).__init__(*args, **kwargs)
|
|
|
|
def validate(self, value):
|
|
if self.multi == False:
|
|
return isinstance(value, str)
|
|
else:
|
|
for val in value:
|
|
if not isinstance(val, str):
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
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 __init__(self, *args, **kwargs):
|
|
super(IPOption, self).__init__(*args, **kwargs)
|
|
|
|
def validate(self, value):
|
|
# by now the validation is nothing but a string, use IPy instead
|
|
if self.multi == False:
|
|
return isinstance(value, str)
|
|
else:
|
|
for val in value:
|
|
if not isinstance(val, str):
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
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 __init__(self, *args, **kwargs):
|
|
super(NetmaskOption, self).__init__(*args, **kwargs)
|
|
|
|
def validate(self, value):
|
|
# by now the validation is nothing but a string, use IPy instead
|
|
if self.multi == False:
|
|
return isinstance(value, str)
|
|
else:
|
|
for val in value:
|
|
if not isinstance(val, str):
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
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])()
|
|
|