from .autopath import do_autopath do_autopath() import warnings from py.test import raises from tiramisu import BoolOption, StrOption, OptionDescription, MasterSlaves, Config, Params, ParamValue, ParamOption, ParamContext from tiramisu.setting import groups from tiramisu.error import ValueWarning, ConfigError from tiramisu.i18n import _ from tiramisu.storage import list_sessions def teardown_function(function): assert list_sessions() == [], 'session list is not empty when leaving "{}"'.format(function.__name__) msg_err = _('attention, "{0}" could be an invalid {1} for "{2}", {3}') def return_true(value, param=None): if value == 'val' and param in [None, 'yes']: return True raise ValueError('test error') def return_false(value, param=None): if value == 'val' and param in [None, 'yes']: raise ValueError('test error return_false') def return_val(value, param=None): return 'val' def return_if_val(value): if value != 'val': raise ValueError('test error') def is_context(value, context): api = Config(context) api.property.pop('validator') if not isinstance(api, Config): raise ValueError('not context') def value_values(value, values): if not (value == 'val' and values == ['val'] or value == 'val1' and values == ['val'] or value == 'val2' and values == ['val'] or value == 'val1' and values == ['val1'] or value == 'val1' and values == ['val1', 'val2'] or value == 'val2' and values == ['val1', 'val2'] or value == 'val1' and values == ['val1', None]): raise ValueError('error') def value_values_index(value, values, index): value_values(value, values) if not (index == 0 or (value == 'val2' and index == 1)): raise ValueError('error 2') def value_values_auto(value, values, auto=False): if auto != False: raise ValueError('auto should be False') if not (value == 'val' and values == ['val'] or value == 'val1' and values == ['val1'] or value == 'val2' and values == ['val1', 'val2'] or value == 'val1' and values == ['val1', None]): raise ValueError('error') def value_values_auto2(value, values, auto=False): if auto != False: raise ValueError('auto should be False') if not (value == 'val1' and values == 'val' or value == 'val2' and values == 'val'): raise ValueError('error') def value_values_index2(value, values, index, auto=False): if auto != False: raise ValueError('auto should be False') if not (value == 'val1' and values == ['val1'] and index == 'val' or value == 'val1' and values == ['val1', None] and index == 'val' or value == 'val2' and values == ['val1', 'val2'] and index == 'val'): raise ValueError('error') def value_empty(value, empty, values): if not value == 'val' or empty is not False and not values == ['val']: raise ValueError('error') def valid_from_config(value, config): api = Config(config) if api.option('opt1').value.get() != u'yes': raise ValueError("c'est une erreur") def test_validator(): opt1 = StrOption('opt1', '', validator=return_true, default='val') raises(ValueError, "StrOption('opt2', '', validator=return_false, default='val')") opt2 = StrOption('opt2', '', validator=return_false) root = OptionDescription('root', '', [opt1, opt2]) api = Config(root) assert api.option('opt1').value.get() == 'val' raises(ValueError, "api.option('opt2').value.set('val')") try: api.option('opt2').value.set('val') except ValueError as err: msg = _('"{0}" is an invalid {1} for "{2}", {3}').format('val', _('string'), 'opt2', 'test error return_false') assert str(err) == msg api.property.add('demoting_error_warning') warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('opt2').value.set('val') assert len(w) == 1 assert str(w[0].message) == msg def test_validator_params(): opt1 = StrOption('opt1', '', validator=return_true, validator_params=Params(ParamValue('yes')), default='val') raises(ValueError, "StrOption('opt2', '', validator=return_false, validator_params=Params(ParamValue('yes')), default='val')") opt2 = StrOption('opt2', '', validator=return_false, validator_params=Params(ParamValue('yes'))) root = OptionDescription('root', '', [opt1, opt2]) api = Config(root) assert api.option('opt1').value.get() == 'val' raises(ValueError, "api.option('opt2').value.set('val')") api.property.add('demoting_error_warning') warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('opt2').value.set('val') assert len(w) == 1 def test_validator_params_value_values(): opt1 = StrOption('opt1', '', validator=value_values, default=['val'], multi=True) root = OptionDescription('root', '', [opt1]) api = Config(root) assert api.option('opt1').value.get() == ['val'] api.option('opt1').value.set(['val1', 'val2']) def test_validator_params_value_values_index(): opt1 = StrOption('opt1', '', validator=value_values_index, default=['val'], multi=True) root = OptionDescription('root', '', [opt1]) api = Config(root) assert api.option('opt1').value.get() == ['val'] api.option('opt1').value.set(['val1', 'val2']) def test_validator_params_value_values_master(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True, validator=value_values) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == [] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val2']) def test_validator_params_value_values_index_master(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True, validator=value_values_index) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == [] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val2']) def test_validator_params_value_values_slave(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validator=value_values) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == [] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val']) api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1') api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val']) api.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2') def test_validator_params_value_values_index_slave(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validator=value_values_index) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == [] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val']) api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1') api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val']) api.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2') def test_validator_params_value_values_notmulti(): raises(ConfigError, "opt1 = StrOption('opt1', '', validator=value_values, default='val')") def test_validator_params_value_values_kwargs_empty(): v = BoolOption('v', '', default=False) ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True, default=["ip"]) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validator=value_empty, validator_params=Params(ParamOption(v))) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [v, interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == ['ip'] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['ip', 'val']) #cfg.ip_admin_eth0.ip_admin_eth0.append('val') #cfg.ip_admin_eth0.netmask_admin_eth0[1] = 'val2' def test_validator_params_value_values_kwargs(): v = BoolOption('v', '', default=False) ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True, default=["ip"]) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validator=value_values_auto, validator_params=Params(kwargs={'auto': ParamOption(v)})) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [v, interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == ['ip'] api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1') api.option('ip_admin_eth0.ip_admin_eth0').value.set(['ip', 'val']) api.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2') def test_validator_params_value_values_kwargs_values(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validator=value_values_auto2, validator_params=Params(kwargs={'values': ParamOption(ip_admin_eth0)})) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == [] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val']) api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1') api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val']) api.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2') def test_validator_params_value_values_kwargs2(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validator=value_values_index2, validator_params=Params(ParamValue(['val1']), {'index': ParamOption(ip_admin_eth0)})) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == [] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val']) api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1') api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val']) def test_validator_params_value_values_kwargs_index(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validator=value_values_index2, validator_params=Params(kwargs={'index': ParamOption(ip_admin_eth0)})) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == [] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val']) api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1') api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val']) api.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2') def test_validator_params_context(): opt1 = StrOption('opt1', '', validator=is_context, validator_params=Params(ParamContext()), default='val') root = OptionDescription('root', '', [opt1]) api = Config(root) assert 'validator' in api.property.get() assert api.option('opt1').value.get() == 'val' assert 'validator' in api.property.get() def test_validator_params_context_value(): opt1 = StrOption('opt1', '', 'yes') opt2 = StrOption('opt2', '', validator=valid_from_config, validator_params=Params(ParamContext()), default='val') root = OptionDescription('root', '', [opt1, opt2]) api = Config(root) assert api.option('opt1').value.get() == 'yes' assert api.option('opt2').value.get() == 'val' api.option('opt1').value.set('no') raises(ValueError, "assert api.option('opt2').value.get()") api.property.add('demoting_error_warning') warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('opt2').value.get() assert len(w) == 1 def test_validator_params_key(): opt1 = StrOption('opt1', '', validator=return_true, validator_params=Params(kwargs={'param': ParamValue('yes')}), default='val') raises(ConfigError, "StrOption('opt2', '', validator=return_true, validator_params=Params(kwargs={'param_unknown': ParamValue('yes')}), default='val')") root = OptionDescription('root', '', [opt1]) api = Config(root) assert api.option('opt1').value.get() == 'val' def test_validator_params_option(): opt0 = StrOption('opt0', '', default='yes') opt1 = StrOption('opt1', '', validator=return_true, validator_params=Params(ParamOption(opt0)), default='val') r = OptionDescription('root', '', [opt0, opt1]) api = Config(r) assert api.option('opt1').value.get() == 'val' api.option('opt0').value.set('val') raises(ValueError, "api.option('opt1').value.get()") api.property.add('demoting_error_warning') warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('opt1').value.get() assert len(w) == 1 def test_validator_multi(): opt1 = StrOption('opt1', '', validator=return_if_val, multi=True) root = OptionDescription('root', '', [opt1]) api = Config(root) assert api.option('opt1').value.get() == [] api.option('opt1').value.set(['val']) assert api.option('opt1').value.get() == ['val'] raises(ValueError, "api.option('opt1').value.set(['val', 'val1'])") api.property.add('demoting_error_warning') warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('opt1').value.set(['val', 'val1']) assert len(w) == 1 def test_validator_warning(): opt1 = StrOption('opt1', '', validator=return_true, default='val', warnings_only=True) opt2 = StrOption('opt2', '', validator=return_false, warnings_only=True) opt3 = StrOption('opt3', '', validator=return_if_val, multi=True, warnings_only=True) root = OptionDescription('root', '', [opt1, opt2, opt3]) api = Config(root) assert api.option('opt1').value.get() == 'val' warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('opt1').value.set('val') assert w == [] # with warnings.catch_warnings(record=True) as w: api.option('opt2').value.set('val') assert len(w) == 1 assert w[0].message.opt() == opt2 assert str(w[0].message) == msg_err.format('val', opt2._display_name, 'opt2', 'test error return_false') # with warnings.catch_warnings(record=True) as w: api.option('opt3').value.set(['val']) assert w == [] # with warnings.catch_warnings(record=True) as w: api.option('opt3').value.set(['val', 'val1']) assert len(w) == 1 assert w[0].message.opt() == opt3 assert str(w[0].message) == msg_err.format('val1', opt3._display_name, 'opt3', 'test error') # with warnings.catch_warnings(record=True) as w: raises(ValueError, "api.option('opt2').value.set(1)") assert len(w) == 0 # with warnings.catch_warnings(record=True) as w: api.option('opt2').value.set('val') api.option('opt3').value.set(['val', 'val1', 'val']) assert len(w) == 2 assert w[0].message.opt() == opt2 assert str(w[0].message) == msg_err.format('val', opt2._display_name, 'opt2', 'test error return_false') assert w[1].message.opt() == opt3 assert str(w[1].message) == msg_err.format('val1', opt3._display_name, 'opt3', 'test error') def test_validator_warning_disabled(): opt1 = StrOption('opt1', '', validator=return_true, default='val', warnings_only=True) opt2 = StrOption('opt2', '', validator=return_false, warnings_only=True) opt3 = StrOption('opt3', '', validator=return_if_val, multi=True, warnings_only=True) root = OptionDescription('root', '', [opt1, opt2, opt3]) api = Config(root) api.property.pop('warnings') assert api.option('opt1').value.get() == 'val' warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('opt1').value.set('val') assert w == [] # with warnings.catch_warnings(record=True) as w: api.option('opt2').value.set('val') assert w == [] # with warnings.catch_warnings(record=True) as w: api.option('opt3').value.set(['val']) assert w == [] # with warnings.catch_warnings(record=True) as w: api.option('opt3').value.set(['val', 'val1']) assert w == [] raises(ValueError, "api.option('opt2').value.set(1)") # with warnings.catch_warnings(record=True) as w: api.option('opt2').value.set('val') api.option('opt3').value.set(['val', 'val1', 'val']) assert w == [] # api.property.add('demoting_error_warning') warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('opt2').value.set(1) assert len(w) == 1 def test_validator_warning_master_slave(): display_name_ip = "ip reseau autorise" display_name_netmask = "masque du sous-reseau" ip_admin_eth0 = StrOption('ip_admin_eth0', display_name_ip, multi=True, validator=return_false, warnings_only=True) netmask_admin_eth0 = StrOption('netmask_admin_eth0', display_name_netmask, multi=True, validator=return_if_val, warnings_only=True) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) assert interface1.impl_get_group_type() == groups.master root = OptionDescription('root', '', [interface1]) api = Config(root) warnings.simplefilter("always", ValueWarning) with warnings.catch_warnings(record=True) as w: api.option('ip_admin_eth0.ip_admin_eth0').value.set([None]) assert w == [] # with warnings.catch_warnings(record=True) as w: api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1') assert len(w) == 1 assert w[0].message.opt() == netmask_admin_eth0 assert str(w[0].message) == msg_err.format('val1', netmask_admin_eth0._display_name, display_name_netmask, 'test error') # with warnings.catch_warnings(record=True) as w: api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val']) assert len(w) == 1 assert w[0].message.opt() == ip_admin_eth0 assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip, 'test error return_false') # with warnings.catch_warnings(record=True) as w: api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val1', 'val1']) #FIXME #assert len(w) == 1 assert w[0].message.opt() == ip_admin_eth0 assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip, 'test error return_false') # with warnings.catch_warnings(record=True) as w: api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val', 'val1']) #FIXME #assert len(w) == 1 assert w[0].message.opt() == ip_admin_eth0 assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip, 'test error return_false') # warnings.resetwarnings() with warnings.catch_warnings(record=True) as w: api.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val1', 'val']) #FIXME #assert len(w) == 1 assert w[0].message.opt() == ip_admin_eth0 assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip, 'test error return_false') def test_validator_slave_param(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise", multi=True) netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validator=return_true, validator_params=Params(kwargs={'param': ParamOption(ip_admin_eth0)})) interface1 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) root = OptionDescription('root', '', [interface1]) api = Config(root) assert api.option('ip_admin_eth0.ip_admin_eth0').value.get() == [] api.option('ip_admin_eth0.ip_admin_eth0').value.set(['yes']) api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val') api.option('ip_admin_eth0.ip_admin_eth0').value.set(['yes', 'yes']) api.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val') def test_validator_dependencies(): ip_admin_eth0 = StrOption('ip_admin_eth0', "ip reseau autorise") netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", validator=return_true, validator_params=Params(kwargs={'param': ParamOption(ip_admin_eth0)})) opt2 = StrOption('opt2', '', validator=return_false) root = OptionDescription('root', '', [ip_admin_eth0, netmask_admin_eth0, opt2]) api = Config(root) assert api.option('ip_admin_eth0').option.has_dependency() is False assert api.option('netmask_admin_eth0').option.has_dependency() is True assert api.option('opt2').option.has_dependency() is False # assert api.option('ip_admin_eth0').option.has_dependency(False) is True assert api.option('netmask_admin_eth0').option.has_dependency(False) is False assert api.option('opt2').option.has_dependency(False) is False