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
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={'': ('yes',)}, default='val')
    raises(ValueError, "StrOption('opt2', '', validator=return_false, validator_params={'': ('yes',)}, default='val')")
    opt2 = StrOption('opt2', '', validator=return_false, validator_params={'': ('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={'': ((v, False),)})
    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={'auto': ((v, False),)})
    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={'values': ((ip_admin_eth0, False),)})
    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={'': (['val1'],), 'index': ((ip_admin_eth0, False),)})
    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={'index': ((ip_admin_eth0, False),)})
    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={'': ((None,),)}, 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={'': ((None,),)}, 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={'param': ('yes',)}, default='val')
    raises(ConfigError, "StrOption('opt2', '', validator=return_true, validator_params={'param_unknown': ('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={'': ((opt0, False),)}, 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={'param': ((ip_admin_eth0, False),)})
    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={'param': ((ip_admin_eth0, False),)})
    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