tiramisu/test/test_option_validator.py

from .autopath import do_autopath
do_autopath()

import warnings
from py.test import raises

from tiramisu import BoolOption, StrOption, OptionDescription, MasterSlaves, Config, getapi, Params, ParamValue, ParamOption, ParamContext
from tiramisu.setting import groups
from tiramisu.error import ValueWarning, ConfigError
from tiramisu.i18n import _


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 = getapi(context)
    api.property.pop('validator')
    if not isinstance(context, 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 = getapi(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 = getapi(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


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 = getapi(Config(root))
    assert api.option('opt1').value.get() == 'val'
    raises(ValueError, "api.option('opt2').value.set('val')")


def test_validator_params_value_values():
    opt1 = StrOption('opt1', '', validator=value_values, default=['val'], multi=True)
    root = OptionDescription('root', '', [opt1])
    api = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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 = getapi(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()")


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 = getapi(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 = getapi(Config(r))
    assert api.option('opt1').value.get() == 'val'
    api.option('opt0').value.set('val')
    raises(ValueError, "api.option('opt1').value.get()")


def test_validator_multi():
    opt1 = StrOption('opt1', '', validator=return_if_val, multi=True)
    root = OptionDescription('root', '', [opt1])
    api = getapi(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'])")


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 = getapi(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 = getapi(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 == []


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 = getapi(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 = getapi(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 = getapi(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