from tiramisu.setting import owners
from tiramisu.option import ChoiceOption, StrOption, OptionDescription
from tiramisu.config import Config
+from tiramisu.error import ConfigError
from py.test import raises
assert cfg.getowner(ch) == owners.default
#
raises(ValueError, "cfg.ch2")
+
+
+def test_choiceoption_calc_invalid():
+ st = StrOption('st', '', ['val1'], multi=True)
+ st
+ raises(ValueError, "ch = ChoiceOption('ch', '', default_multi='val2', values=[1, 2, 3], values_params={'': ((st, False),)}, multi=True)")
+
+
+def test_choiceoption_calc_not_list():
+ st = StrOption('st', '', 'val1')
+ ch = ChoiceOption('ch', "", default_multi='val2', values=return_val, values_params={'': ((st, False),)}, multi=True)
+ od = OptionDescription('od', '', [st, ch])
+ cfg = Config(od)
+ cfg.read_write()
+ raises(ConfigError, "cfg.ch = ['val1']")
assert dm.impl_getname() == 'dummy'
+def test_base_config_name():
+ gcdummy = BoolOption('dummy', 'dummy', default=False)
+ descr = OptionDescription('tiramisu', '', [gcdummy])
+ cfg = Config(descr, name='cfg')
+ cfg.impl_getname() == 'cfg'
+ raises(ValueError, "Config(descr, name='unvalid name')")
+
+
def test_not_config():
assert raises(TypeError, "Config('str')")
raises(ConfigError, 'sub.make_dict()')
+def test_subconfig():
+ i = IntOption('i', '')
+ o = OptionDescription('val', '', [i])
+ o2 = OptionDescription('val', '', [o])
+ c = Config(o2)
+ raises(TypeError, "SubConfig(i, weakref.ref(c))")
+ pass
+
+
def test_config_weakref():
o = OptionDescription('val', '', [])
o2 = OptionDescription('val', '', [o])
raises(ValueError, "c.f = 'domainnametoolongthathavemorethanmaximumsizeforatruedomainnamean'")
c.f = 'domainnametoolongthathavemorethanmaximumsizeforatruedomainnamea.nd'
c.f = 'domainnametoolongthathavemorethanmaximumsizeforatruedomainnamea.nditsnoteasytogeneratesolongdomainnamewithoutrepeatdomainnameto.olongthathavemorethanmaximumsizeforatruedomainnameanditsnoteas.ytogeneratesolongdomainnamewithoutrepeatbutimnotabletodoitnowie'
- raises(ValueError, "c.f = 'domainnametoolongthathavemorethanmaximumsizeforatruedomainnamea.nditsnoteasytogeneratesolongdomainnamewithoutrepeatdomainnameto.olongthathavemorethanmaximumsizeforatruedomainnameanditsnoteas.ytogeneratesolongdomainnamewithoutrepeatbutimnotabletodoitnowien'")
+ raises(ValueError, "c.f = 'domainnametoolongthathavemorethanmaximumsizeforatruedomainname.nditsnoteasytogeneratesolongdomainnamewithoutrepeatdomainnamet.olongthathavemorethanmaximumsizeforatruedomainnameanditsnotea.ytogeneratesolongdomainnamewithoutrepeatbutimnotabletodoitnowie.xxxx'")
c.f = 'd'
c.f = 'd.t'
#
from autopath import do_autopath
do_autopath()
+import warnings
from py.test import raises
from tiramisu.config import Config
from tiramisu.option import IPOption, NetworkOption, NetmaskOption, \
PortOption, OptionDescription
+from tiramisu.error import ValueWarning
def test_ip():
a = IPOption('a', '')
b = IPOption('b', '', private_only=True)
- od = OptionDescription('od', '', [a, b])
+ d = IPOption('d', '', warnings_only=True, private_only=True)
+ warnings.simplefilter("always", ValueWarning)
+ od = OptionDescription('od', '', [a, b, d])
c = Config(od)
c.a = '192.168.1.1'
c.a = '192.168.1.0'
raises(ValueError, "c.b = '255.255.255.0'")
raises(ValueError, "IPOption('a', 'ip', default='192.000.023.01')")
- d = IPOption('a', 'ip', default='192.0.23.1')
- od = OptionDescription('od', '', [d])
- c = Config(od)
- raises(ValueError, "c.a = '192.000.023.01'")
+ with warnings.catch_warnings(record=True) as w:
+ c.d = '88.88.88.88'
+ assert len(w) == 1
def test_ip_default():
def test_ip_reserved():
a = IPOption('a', '')
b = IPOption('b', '', allow_reserved=True)
- od = OptionDescription('od', '', [a, b])
- c = Config(od)
- raises(ValueError, "c.a = '226.94.1.1'")
- c.b = '226.94.1.1'
+ c = IPOption('c', '', warnings_only=True)
+ od = OptionDescription('od', '', [a, b, c])
+ warnings.simplefilter("always", ValueWarning)
+ cfg = Config(od)
+ raises(ValueError, "cfg.a = '226.94.1.1'")
+ cfg.b = '226.94.1.1'
+ with warnings.catch_warnings(record=True) as w:
+ cfg.c = '226.94.1.1'
+ assert len(w) == 1
def test_network():
a = NetworkOption('a', '')
- od = OptionDescription('od', '', [a])
+ b = NetworkOption('b', '', warnings_only=True)
+ od = OptionDescription('od', '', [a, b])
+ warnings.simplefilter("always", ValueWarning)
c = Config(od)
c.a = '192.168.1.1'
c.a = '192.168.1.0'
c.a = '88.88.88.88'
c.a = '0.0.0.0'
raises(ValueError, "c.a = '255.255.255.0'")
+ with warnings.catch_warnings(record=True) as w:
+ c.b = '255.255.255.0'
+ assert len(w) == 1
def test_netmask():
raises(ValueError, "cfg.od.dodval2.stval2 = 'yes'")
+def test_consistency_dyndescription_default():
+ st = StrOption('st', '', 'yes')
+ st2 = StrOption('st2', '')
+ dod = DynOptionDescription('dod', '', [st, st2], callback=return_list)
+ od = OptionDescription('od', '', [dod])
+ st.impl_add_consistency('not_equal', st2)
+ od2 = OptionDescription('od', '', [od])
+ cfg = Config(od2)
+ cfg
+ raises(ValueError, "cfg.od.dodval1.st2val1 = 'yes'")
+ raises(ValueError, "cfg.od.dodval2.st2val2 = 'yes'")
+
+
+def test_consistency_dyndescription_multi():
+ st = StrOption('st', '', multi=True)
+ st2 = StrOption('st2', '', multi=True)
+ dod = DynOptionDescription('dod', '', [st, st2], callback=return_list)
+ od = OptionDescription('od', '', [dod])
+ st.impl_add_consistency('not_equal', st2)
+ od2 = OptionDescription('od', '', [od])
+ cfg = Config(od2)
+ cfg.od.dodval1.stval1.append('yes')
+ raises(ValueError, "cfg.od.dodval1.st2val1.append('yes')")
+ cfg.od.dodval2.stval2.append('yes')
+ raises(ValueError, "cfg.od.dodval2.st2val2.append('yes')")
+ raises(ValueError, "cfg.od.dodval1.st2val1.append('yes')")
+ del(cfg.od.dodval2.stval2)
+ raises(ValueError, "cfg.od.dodval1.st2val1.append('yes')")
+ cfg.od.dodval2.st2val2.append('yes')
+ raises(ValueError, "cfg.od.dodval2.stval2.append('yes')")
+
+
+def test_consistency_dyndescription_default_multi():
+ st = StrOption('st', '', ['yes'], multi=True)
+ st2 = StrOption('st2', '', multi=True)
+ dod = DynOptionDescription('dod', '', [st, st2], callback=return_list)
+ od = OptionDescription('od', '', [dod])
+ st.impl_add_consistency('not_equal', st2)
+ od2 = OptionDescription('od', '', [od])
+ cfg = Config(od2)
+ raises(ValueError, "cfg.od.dodval1.st2val1.append('yes')")
+ raises(ValueError, "cfg.od.dodval1.st2val1.append('yes')")
+ cfg.od.dodval1.stval1.append('yes')
+
+
+def test_consistency_dyndescription_default_multi2():
+ st = StrOption('st', '', ['yes'], multi=True)
+ st2 = StrOption('st2', '', ['yes'], multi=True)
+ dod = DynOptionDescription('dod', '', [st, st2], callback=return_list)
+ dod
+ raises(ValueError, "st.impl_add_consistency('not_equal', st2)")
+
+
+def test_consistency_only_one_dyndescription():
+ st = StrOption('st', '')
+ st
+ st2 = StrOption('st2', '')
+ DynOptionDescription('dod', '', [st2], callback=return_list)
+ raises(ConfigError, "st.impl_add_consistency('not_equal', st2)")
+ raises(ConfigError, "st2.impl_add_consistency('not_equal', st)")
+
+
+def test_consistency_became_dyndescription():
+ st = StrOption('st', '')
+ st2 = StrOption('st2', '')
+ st2.impl_add_consistency('not_equal', st)
+ od = DynOptionDescription('dod', '', [st2], callback=return_list)
+ od2 = OptionDescription('od', '', [od, st])
+ od2
+ raises(ConfigError, "c = Config(od2)")
+
+
+def test_consistency_became_dyndescription2():
+ st = StrOption('st', '')
+ st2 = StrOption('st2', '')
+ st.impl_add_consistency('not_equal', st2)
+ od = DynOptionDescription('dod', '', [st2], callback=return_list)
+ od2 = OptionDescription('od', '', [od, st])
+ od2
+ raises(ConfigError, "c = Config(od2)")
+
+
def test_consistency_external_dyndescription():
st = StrOption('st', '')
st1 = StrOption('st1', '')
conf2 = Config(od2, name='conf2')
conf1.read_write()
conf2.read_write()
- meta = MetaConfig([conf1, conf2])
+ meta = MetaConfig([conf1, conf2], name='meta')
meta.cfgimpl_get_settings().setowner(owners.meta)
return meta
raises(AttributeError, "meta2.find_firsts(byname='i1', byvalue=10)")
raises(AttributeError, "meta2.find_firsts(byname='not', byvalue=10)")
raises(AttributeError, "meta2.find_firsts(byname='i6')")
+ raises(ValueError, "meta2.set_value('od1.i6', 7, only_config=True, force_default=True)")
+ raises(ValueError, "meta2.set_value('od1.i6', 7, only_config=True, force_default_if_same=True)")
+ raises(ValueError, "meta2.set_value('od1.i6', 7, only_config=True, force_dont_change_value=True)")
def test_not_meta():
assert [conf1, conf2] == grp.find_firsts(byname='i1').cfgimpl_get_children()
+def test_group_group():
+ i1 = IntOption('i1', '')
+ od1 = OptionDescription('od1', '', [i1])
+ od2 = OptionDescription('od2', '', [od1])
+ conf1 = Config(od2, name='conf1')
+ conf2 = Config(od2, name='conf2')
+ grp = GroupConfig([conf1, conf2])
+ raises(ValueError, "GroupConfig([grp])")
+ grp = GroupConfig([conf1, conf2], 'grp')
+ grp2 = GroupConfig([grp])
+ grp2.set_value('od1.i1', 2)
+ assert grp2.grp.conf1.od1.i1 == 2
+ assert grp2.grp.conf1.getowner(i1) == owners.user
+
+
def test_meta_path():
meta = make_description()
assert meta._impl_path is None
assert cfg.od.st.st1.st1 == ['yes']
assert cfg.od.st.st1.st2 == ['yes']
assert cfg.getowner(st1) == owner
+
+
+def test_re_set_callback():
+ st1 = StrOption('st1', "", multi=True)
+ st2 = StrOption('st2', "", multi=True)
+ st2.impl_set_callback(return_value, {'': ((st1, False),)})
+ raises(ConfigError, "st2.impl_set_callback(return_value, {'': ((st1, False),)})")
from tiramisu.setting import owners, groups
from tiramisu.config import Config
from tiramisu.option import IPOption, NetworkOption, NetmaskOption, IntOption,\
- BroadcastOption, SymLinkOption, OptionDescription
+ BroadcastOption, SymLinkOption, OptionDescription, submulti
from tiramisu.error import ConfigError, ValueWarning, PropertiesOptionError
import warnings
def test_consistency():
a = IntOption('a', '')
b = IntOption('b', '')
- od = OptionDescription('od', '', [a, b])
a.impl_add_consistency('not_equal', b)
#consistency to itself
raises(ConfigError, "a.impl_add_consistency('not_equal', a)")
def test_consistency_not_exists():
a = IntOption('a', '')
b = IntOption('b', '')
- od = OptionDescription('od', '', [a, b])
+ a, b
raises(ConfigError, "a.impl_add_consistency('not_exists', b)")
def test_consistency_unknown_params():
a = IntOption('a', '')
b = IntOption('b', '')
- od = OptionDescription('od', '', [a, b])
+ a, b
raises(ValueError, "a.impl_add_consistency('not_equal', b, unknown=False)")
+def test_consistency_warnings_only_default():
+ a = IntOption('a', '', 1)
+ b = IntOption('b', '', 1)
+ warnings.simplefilter("always", ValueWarning)
+ with warnings.catch_warnings(record=True) as w:
+ a.impl_add_consistency('not_equal', b, warnings_only=True)
+ assert w != []
+
+
def test_consistency_warnings_only():
a = IntOption('a', '')
b = IntOption('b', '')
b = IntOption('b', '')
a.impl_add_consistency('not_equal', b)
od1 = OptionDescription('od1', '', [a])
- od2 = OptionDescription('od2', '', [b])
od = OptionDescription('root', '', [od1])
+ od
raises(ConfigError, "Config(od)")
od1 = OptionDescription('od1', '', [a])
od2 = OptionDescription('od2', '', [b])
od = OptionDescription('root', '', [od1, od2])
+ od
#with subconfig
raises(ConfigError, "Config(od.od1)")
assert set(od._cache_consistencies.keys()) == set([a, b])
+def test_consistency_not_equal_submulti():
+ a = IntOption('a', '', multi=submulti)
+ b = IntOption('b', '', multi=submulti)
+ od = OptionDescription('od', '', [a, b])
+ a.impl_add_consistency('not_equal', b)
+ c = Config(od)
+ assert c.a == []
+ assert c.b == []
+ c.a = [[1]]
+ del(c.a)
+ c.a = [[1]]
+ raises(ValueError, "c.b = [[1]]")
+ c.a = [[1, 2]]
+ c.b = [[3]]
+ c.b = [[3, 1]]
+ c.b = [[3]]
+ c.b[0].append(1)
+ c.b = [[3], [1]]
+
+
+def test_consistency_not_equal_default_submulti():
+ a = IntOption('a', '', [[1, 2]], multi=submulti)
+ b = IntOption('b', '', [[1]], multi=submulti)
+ od = OptionDescription('od', '', [a, b])
+ od
+ raises(ValueError, "a.impl_add_consistency('not_equal', b)")
+
+
def test_consistency_not_equal_multi():
a = IntOption('a', '', multi=True)
b = IntOption('b', '', multi=True)
c.b = [2]
+def test_consistency_not_equal_multi_default():
+ a = IntOption('a', '', multi=True)
+ b = IntOption('b', '', multi=True, default_multi=1)
+ od = OptionDescription('od', '', [a, b])
+ a.impl_add_consistency('not_equal', b)
+ c = Config(od)
+ assert c.a == []
+ assert c.b == []
+ c.a = [1]
+ del(c.a)
+ c.a = [1]
+ raises(ValueError, "c.b = [1]")
+ c.b = [2]
+
+
def test_consistency_default():
a = IntOption('a', '', 1)
b = IntOption('b', '', 1)
+ a, b
raises(ValueError, "a.impl_add_consistency('not_equal', b)")
a = IntOption('a', '', [2, 1], multi=True)
b = IntOption('b', '', [1, 1], multi=True)
c = IntOption('c', '', [1, 2], multi=True)
+ b
raises(ValueError, "a.impl_add_consistency('not_equal', b)")
a.impl_add_consistency('not_equal', c)
a = NetworkOption('a', '')
b = NetmaskOption('b', '')
c = IPOption('c', '')
- od = OptionDescription('od', '', [a, b, c])
+ d = IPOption('d', '')
+ od = OptionDescription('od', '', [a, b, c, d])
c.impl_add_consistency('in_network', a, b)
+ d.impl_add_consistency('in_network', a, b, warnings_only=True)
+ warnings.simplefilter("always", ValueWarning)
cfg = Config(od)
cfg.a = '192.168.1.0'
cfg.b = '255.255.255.0'
raises(ValueError, "cfg.c = '192.168.2.1'")
raises(ValueError, "cfg.c = '192.168.1.0'")
raises(ValueError, "cfg.c = '192.168.1.255'")
+ with warnings.catch_warnings(record=True) as w:
+ cfg.d = '192.168.2.1'
+ assert len(w) == 1
def test_consistency_ip_in_network_len_error():
od = OptionDescription('od', '', [a, b, c])
c.impl_add_consistency('in_network', a)
cfg = Config(od)
+ cfg
raises(ConfigError, "cfg.a = '192.168.2.0'")
def test_consistency_ip_netmask_error_multi():
a = IPOption('a', '', multi=True)
b = NetmaskOption('b', '')
- od = OptionDescription('od', '', [a, b])
+ OptionDescription('od', '', [a, b])
raises(ConfigError, "b.impl_add_consistency('ip_netmask', a)")
a = NetworkOption('a', '', '192.168.1.0')
b = NetmaskOption('b', '', '255.255.255.128')
c = BroadcastOption('c', '', '192.168.2.127')
- d = BroadcastOption('d', '', '192.168.1.127')
od = OptionDescription('a', '', [a, b, c])
+ od
raises(ValueError, "c.impl_add_consistency('broadcast', a, b)")
def test_consistency_broadcast_default_2():
a = NetworkOption('a', '', '192.168.1.0')
b = NetmaskOption('b', '', '255.255.255.128')
- c = BroadcastOption('c', '', '192.168.2.127')
d = BroadcastOption('d', '', '192.168.1.127')
od2 = OptionDescription('a', '', [a, b, d])
+ od2
d.impl_add_consistency('broadcast', a, b)
from py.test import raises
-def return_val():
- return 'val'
+def return_val(val=None):
+ if val is None:
+ return 'val'
+ else:
+ return val
def return_list(value=None):
cfg.multi.append()
assert cfg.getowner(multi) == owner
assert cfg.multi == [['val', 'val'], ['val', 'val']]
+ cfg.multi.append()
+ assert cfg.multi == [['val', 'val'], ['val', 'val'], ['val', 'val']]
del(cfg.multi)
assert cfg.getowner(multi) == owners.default
cfg.multi[0].append()
cfg.multi[0].append()
assert cfg.getowner(multi) == owner
assert cfg.multi == [['val', 'val', None]]
+ del(cfg.multi)
+ cfg.multi.append()
#FIXME multi sur une master
cfg.ip_admin_eth0.ip_admin_eth0[0].pop(0)
assert cfg.ip_admin_eth0.ip_admin_eth0 == [["192.168.1.1"], ["192.168.230.147"]]
raises(ValueError, 'cfg.ip_admin_eth0.ip_admin_eth0 = ["192.168.1.1", "192.168.1.1"]')
+
+
+def test_callback_submulti():
+ multi = StrOption('multi', '', multi=submulti)
+ multi2 = StrOption('multi2', '', multi=submulti, callback=return_val, callback_params={'': ((multi, False),)})
+ od = OptionDescription('multi', '', [multi, multi2])
+ cfg = Config(od)
+ cfg.read_write()
+ owner = cfg.cfgimpl_get_settings().getowner()
+ assert cfg.getowner(multi) == owners.default
+ assert cfg.multi == []
+ assert cfg.multi2 == []
+ cfg.multi.append(['val'])
+ assert cfg.getowner(multi) == owner
+ assert cfg.getowner(multi2) == owners.default
+ assert cfg.multi == [['val']]
+ assert cfg.multi2 == [['val']]
# if callback_params has a callback, launch several time calculate()
master_slave = False
# multi's option should have same value for all option
- len_multi = None
try:
if option._is_subdyn():
tcparams[''] = [(option.impl_getsuffix(), False)]
if opt.impl_is_master_slaves() and \
opt.impl_get_master_slaves().in_same_group(option):
- len_multi = len(value)
master_slave = True
is_multi = True
else:
# if no index, return a list
if master_slave:
ret = []
- if index is not undefined:
- # for example if append master and get a no default slave without
- # getting master
- range_ = [index]
- else:
- range_ = range(len_multi)
- for incr in range_:
- args = []
- kwargs = {}
- for key, couples in tcparams.items():
- for couple in couples:
- value, ismulti = couple
- if ismulti:
- val = value[incr]
- else:
- val = value
- if key == '':
- args.append(val)
- else:
- kwargs[key] = val
- calc = calculate(callback, args, kwargs)
- if index is not undefined:
- ret = calc
- else:
- ret.append(calc)
- return ret
+ args = []
+ kwargs = {}
+ for key, couples in tcparams.items():
+ for couple in couples:
+ value, ismulti = couple
+ if ismulti:
+ val = value[index]
+ else:
+ val = value
+ if key == '':
+ args.append(val)
+ else:
+ kwargs[key] = val
+ return calculate(callback, args, kwargs)
else:
# no value is multi
# return a single value
else:
kwargs[key] = couple[0]
ret = calculate(callback, args, kwargs)
- if callback_params != {}:
- if isinstance(ret, list) and index is not undefined:
- if option.impl_is_master_slaves('slave'):
- raise SlaveError(_("callback cannot return a list for a "
- "slave option ({0})").format(path))
- if len(ret) < index + 1:
- ret = None
- else:
- ret = ret[index]
+ if callback_params != {} and isinstance(ret, list) and index is not undefined:
+ raise SlaveError(_("callback cannot return a list for a "
+ "slave option ({0})").format(path))
return ret
def impl_getname(self):
return self._impl_name
- def impl_setname(self, name):
- self._impl_name = name
-
class GroupConfig(_CommonConfig):
__slots__ = ('__weakref__', '_impl_children', '_impl_name')
for child in children:
if not isinstance(child, _CommonConfig):
raise ValueError(_("groupconfig's children must be Config, MetaConfig or GroupConfig"))
- name = child._impl_name
- if name is None:
+ name_ = child._impl_name
+ if name_ is None:
raise ValueError(_('name must be set to config before creating groupconfig'))
- #if name in names:
- # raise ValueError(_('config name must be uniq in groupconfig'))
- names.append(name)
+ names.append(name_)
if len(names) != len(set(names)):
for idx in xrange(1, len(names) + 1):
name = names.pop(0)
def set_value(self, path, value, force_default=False,
force_dont_change_value=False, force_default_if_same=False,
only_config=False):
+ """only_config: could be set if you want modify value in all Config included in
+ this MetaConfig
+ """
if only_config:
if force_default or force_default_if_same or force_dont_change_value:
raise ValueError(_('force_default, force_default_if_same or '
default = []
self.impl_validate(default)
self._set_default_values(default, default_multi)
+ #callback is False in optiondescription
if callback is not False:
self.impl_set_callback(callback, callback_params)
self.commit()
:param load: `True` if we are at the init of the option description
:type load: bool
"""
- if not load and self.impl_getrequires() is None:
+ if not load and self.impl_getrequires() == []:
self._state_requires = None
elif load and self._state_requires is None:
del(self._state_requires)
if self.__class__.__name__ == 'OptionDescription' or \
isinstance(self, SymLinkOption):
return
- if not load and self.impl_get_callback() is None:
+ 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:
else:
callback, callback_params = self.impl_get_callback()
self._state_callback_params = {}
- if callback_params is not None:
- 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)
- else:
- cllbck_prms = None
+ 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)
except (KeyError, AttributeError):
pass
elif name == '_opt':
- try:
- if self._impl_getopt() is not None:
- #so _opt is already set
- is_readonly = True
- except (KeyError, AttributeError):
- pass
+ pass
elif name != '_readonly':
is_readonly = self.impl_is_readonly()
if is_readonly: # pragma: optional cover
try:
all_cons_vals.append(opt_value[index])
except IndexError, err:
- log.debug('indexerror in _launch_consistency: {0}'.format(err))
#value is not already set, could be higher index
#so return if no value and not 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)
- default_value = context.cfgimpl_get_values()._getvalue(opt, path, True, index=index)
- else:
- default_value = opt.impl_getdefault_multi()
- if default_value is not None:
- all_cons_vals.append(default_value)
- else:
- return
+ 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:
if self._is_warnings_only():
warning = error
error = None
- except ValueWarning as warning:
- log.debug(_('do_validation for {0}: warning in value').format(
- self.impl_getname()), exc_info=True)
- pass
if error is None and warning is None:
try:
# if context launch consistency validation
- if context is not undefined:
- descr._valid_consistency(current_opt, _value, context,
- _index, submulti_index)
+ #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)
if warning:
msg = _("warning on the value of the option {0}: {1}").format(
self.impl_getname(), warning)
- if context is None or 'warnings' in \
+ if context is undefined or 'warnings' in \
context.cfgimpl_get_settings():
warnings.warn_explicit(ValueWarning(msg, self),
ValueWarning,
self.impl_getname(), error))
# generic calculation
- if context is not undefined:
- descr = context.cfgimpl_get_description()
+ #if context is not undefined:
+ # descr = context.cfgimpl_get_description()
if not self.impl_is_multi():
do_validation(value, None, None)
"accesses the Option's doc"
return self.impl_get_information('doc')
- #def impl_getkey(self, value):
- # return value
-
- 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()))
- warnings_only = False
- transitive = True
- for key, value in params.items():
- if key == 'warnings_only':
- warnings_only = value
- elif key == 'transitive':
- transitive = value
- else:
- raise ValueError(_('unknow parameter {0} in consistency').format(key))
+ def _valid_consistencies(self, other_opts):
if self._is_subdyn():
dynod = self._impl_getsubdyn()
else:
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))
- value = self.impl_getdefault()
- if value is not None:
- if self.impl_is_multi():
- for idx, val in enumerate(value):
- if not self.impl_is_submulti():
- self._launch_consistency(func, self, val, undefined, idx,
- None, all_cons_opts,
- warnings_only, transitive)
- else:
- for slave_idx, val in enumerate(value):
- self._launch_consistency(func, self, val, None,
- idx, slave_idx,
- all_cons_opts,
- warnings_only, transitive)
- else:
- self._launch_consistency(func, self, value, undefined, None,
- None, all_cons_opts, warnings_only,
- transitive)
+ 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)
- #re validate default value when add consistency
- self.impl_validate(self.impl_getdefault())
+ #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):
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()))
:param load: `True` if we are at the init of the option description
:type load: bool
"""
- if not load and self._get_consistencies() is None:
+ if not load and self._get_consistencies() == ():
self._state_consistencies = None
elif load and self._state_consistencies is None:
del(self._state_consistencies)
def _validate_callback(self, callback, callback_params):
if callback is None:
return
- try:
- default_multi = self.impl_getdefault_multi()
- except AttributeError:
- default_multi = None
+ default_multi = self.impl_getdefault_multi()
if (not self.impl_is_multi() and (self.impl_getdefault() is not None or
default_multi is not None)) or \
(self.impl_is_multi() and (self.impl_getdefault() != [] or
know more about
the description of the requires dictionary
"""
- if requires is None:
- return None, None
ret_requires = {}
config_action = {}
# sometimes an ip term starts with a zero
# but this does not fit in some case, for example bind does not like it
self._impl_valid_unicode(value)
- try:
- for val in value.split('.'):
- if val.startswith("0") and len(val) > 1:
- raise ValueError(_('invalid IP')) # pragma: optional cover
- except AttributeError: # pragma: optional cover
- #if integer for example
- raise ValueError(_('invalid IP'))
+ for val in value.split('.'):
+ if val.startswith("0") and len(val) > 1:
+ raise ValueError(_('invalid IP')) # pragma: optional cover
# 'standard' validation
try:
IP('{0}/32'.format(value))
def _validate(self, value, context=undefined):
self._impl_valid_unicode(value)
+
def _valid_length(val):
if len(val) < 1:
raise ValueError(_("invalid domainname's length (min 1)"))
from tiramisu.i18n import _
from tiramisu.setting import groups, undefined # , log
-from .baseoption import BaseOption, DynSymLinkOption, SymLinkOption, \
- allowed_character
+from .baseoption import BaseOption, SymLinkOption, allowed_character
from . import MasterSlaves
-from tiramisu.error import ConfigError, ConflictError, ValueWarning
+from tiramisu.error import ConfigError, ConflictError
from tiramisu.storage import get_storages_option
from tiramisu.autolib import carry_out_calculation
cache_option.append(option._get_id())
if not isinstance(option, OptionDescription):
for func, all_cons_opts, params in option._get_consistencies():
+ all_cons_opts[0]._valid_consistencies(all_cons_opts[1:])
for opt in all_cons_opts:
_consistencies.setdefault(opt,
[]).append((func,
raise ValueError(_('group_type: {0}'
' not allowed').format(group_type))
- def _valid_consistency(self, option, value, context, index, submulti_idx):
- if self._cache_consistencies is None:
- return True
- #consistencies is something like [('_cons_not_equal', (opt1, opt2))]
- if isinstance(option, DynSymLinkOption):
- consistencies = self._cache_consistencies.get(option._impl_getopt())
- else:
- consistencies = self._cache_consistencies.get(option)
- 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 _impl_getstate(self, descr=None):
"""enables us to export into a dict
:param descr: parent :class:`tiramisu.option.OptionDescription`
for child in children:
if isinstance(child, OptionDescription):
if child.impl_get_group_type() != groups.master:
- raise ConfigError(_('cannot set optiondescription in an '
+ raise ConfigError(_('cannot set optiondescription in a '
'dynoptiondescription'))
for chld in child._impl_getchildren():
chld._impl_setsubdyn(self)
if isinstance(child, SymLinkOption):
- raise ConfigError(_('cannot set symlinkoption in an '
+ raise ConfigError(_('cannot set symlinkoption in a '
+ 'dynoptiondescription'))
+ if isinstance(child, SymLinkOption):
+ raise ConfigError(_('cannot set symlinkoption in a '
'dynoptiondescription'))
child._impl_setsubdyn(self)
self.impl_set_callback(callback, callback_params)
except AttributeError:
self._consistencies = [cons]
+ def _del_consistency(self):
+ self._consistencies.pop(-1)
+
def _get_consistencies(self):
try:
return self._consistencies
def _get_multi(self, opt, path):
return Multi([], self.context, opt, path)
- def _getdefaultvalue(self, opt, path, with_meta, index):
+ def _getdefaultvalue(self, opt, path, with_meta, index, submulti_index):
# if value has callback and is not set
if opt.impl_has_callback():
callback, callback_params = opt.impl_get_callback()
callback=callback,
callback_params=callback_params,
index=index)
- try:
- if isinstance(value, list) and index is not undefined:
- #if submulti and return a list of list, just return list
- if opt.impl_is_submulti():
- val = value[index]
- if isinstance(val, list):
- value = val
- else:
- value = value[index]
+ if isinstance(value, list) and index is not undefined:
+ #if return a list and index is set, return value only if
+ #it's a submulti without submulti_index and without list of list
+ if opt.impl_is_submulti() and submulti_index is undefined and \
+ (len(value) == 0 or not isinstance(value[0], list)):
+ return value
+ else:
return value
- except IndexError:
- pass
if with_meta:
meta = self._getcontext().cfgimpl_get_meta()
if meta is not None:
# now try to get default value
value = opt.impl_getdefault()
if opt.impl_is_multi() and index is not undefined:
- if value is None:
+ if value == []:
value = opt.impl_getdefault_multi()
else:
try:
value = opt.impl_getdefault_multi()
return value
- def _getvalue(self, opt, path, is_default, index=undefined,
- with_meta=True, self_properties=undefined,
+ def _getvalue(self, opt, path, is_default, self_properties,
+ index=undefined, submulti_index=undefined, with_meta=True,
masterlen=undefined):
"""actually retrieves the value
:param opt: the `option.Option()` object
:returns: the option's value (or the default value if not set)
"""
- if opt.impl_is_optiondescription(): # pragma: optional cover
- raise ValueError(_('optiondescription has no value'))
-
- if self_properties is undefined:
- self_properties = self._getcontext().cfgimpl_get_settings()._getproperties(
- opt, path, read_write=False)
force_default = 'frozen' in self_properties and \
'force_default_on_freeze' in self_properties
+ # not default value
if not is_default and not force_default:
if opt.impl_is_master_slaves('slave'):
- #if masterlen is not undefined:
- if index is undefined:
- value = []
- vals = self._p_.getvalue(path)
- length = max(masterlen, len(vals))
- for idx in xrange(0, length):
- try:
- if vals[idx] is undefined:
- value.append(self._getdefaultvalue(opt, path, with_meta, idx))
- else:
- value.append(vals[idx])
- except IndexError:
- try:
- value.append(self._getdefaultvalue(opt, path, with_meta, idx))
- except IndexError:
- value.append(None)
- else:
- value = self._p_.getvalue(path, index)
- return value
+ return self._p_.getvalue(path, index)
else:
value = self._p_.getvalue(path)
if index is not undefined:
pass
else:
return value
- return self._getdefaultvalue(opt, path, with_meta, index)
+ return self._getdefaultvalue(opt, path, with_meta, index, submulti_index)
def get_modified_values(self):
context = self._getcontext()
self_properties=self_properties,
index=index)
try:
- if index is None:
- gv_index = undefined
- else:
- gv_index = index
- value = self._getvalue(opt, path, is_default, index=gv_index,
+ value = self._getvalue(opt, path, is_default, self_properties,
+ index=index, submulti_index=submulti_index,
with_meta=with_meta,
- self_properties=self_properties,
masterlen=masterlen)
config_error = None
except ConfigError as err:
else:
force_index = index
if opt.impl_is_multi():
- #for slave is a multi
- if index is None and not isinstance(value, list):
- value = []
if force_index is None:
value = Multi(value, self.context, opt, path)
elif opt.impl_is_submulti() and submulti_index is undefined:
" which is a slave").format(self.opt.impl_getname()))
index = self.__len__()
if value is undefined:
- try:
- value = self._get_validated_value(index)
- except IndexError:
- value = None
+ value = self._get_validated_value(index)
context = self._getcontext()
setting = context.cfgimpl_get_settings()
setting_properties = setting._getproperties(read_write=False)
projects / tiramisu.git / commitdiff