tiramisu/tests/test_option_consistency.py

from .autopath import do_autopath
do_autopath()
from .config import config_type, get_config

from py.test import raises

from tiramisu.setting import owners, groups
from tiramisu import IPOption, NetworkOption, NetmaskOption, IntOption,\
    BroadcastOption, StrOption, SymLinkOption, OptionDescription, submulti, Leadership,\
    Config, undefined, Params, ParamOption, Calculation, ParamIndex
from tiramisu.error import ConfigError, ValueWarning, PropertiesOptionError
import warnings
from tiramisu.storage import list_sessions
from tiramisu.i18n import _


def teardown_function(function):
    assert list_sessions() == [], 'session list is not empty when leaving "{}"'.format(function.__name__)


def test_consistency():
    a = IntOption('a', '')
    b = IntOption('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', '')
    a, b
    raises(ConfigError, "a.impl_add_consistency('not_exists', b)")


def test_consistency_unknown_params():
    a = IntOption('a', '')
    b = IntOption('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(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '')
    c = IntOption('c', '')
    od = OptionDescription('od', '', [a, b, c])
    a.impl_add_consistency('not_equal', b, warnings_only=True)
    cfg = Config(od)
    assert cfg.option('a').option.consistencies()
    assert not cfg.option('b').option.consistencies()
    assert not cfg.option('c').option.consistencies()
    cfg = get_config(cfg, config_type)
    cfg.option('a').value.set(1)
    warnings.simplefilter("always", ValueWarning)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('b').value.set(1)
    assert w != []


def test_consistency_warnings_only_more_option(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '')
    d = IntOption('d', '')
    od = OptionDescription('od', '', [a, b, d])
    a.impl_add_consistency('not_equal', b, d, warnings_only=True)
    cfg = Config(od)
    cfg = get_config(cfg, config_type)
    cfg.option('a').value.set(1)
    warnings.simplefilter("always", ValueWarning)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('b').value.set(1)
    assert w != []
    assert len(w) == 1
    with warnings.catch_warnings(record=True) as w:
        cfg.option('d').value.get()
    assert w != []
    assert len(w) == 1
    with warnings.catch_warnings(record=True) as w:
        cfg.option('d').value.set(1)
    assert w != []
    assert len(w) == 1


def test_consistency_error_prefix():
    a = IntOption('a', '')
    b = IntOption('b', '')
    od = OptionDescription('od', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg = Config(od)
    cfg.option('a').value.set(1)
    try:
        cfg.option('b').value.set(1)
    except Exception as err:
        assert str(err) == _('"{0}" is an invalid {1} for "{2}"').format('1', _('integer'), 'b') + ', ' +  _('must be different from the value of {}').format('"a"')
    try:
        cfg.option('b').value.set(1)
    except Exception as err:
        err.prefix = ''
        assert str(err) == _('must be different from the value of {}').format('"a"')


def test_consistency_warnings_only_option(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '', warnings_only=True)
    od = OptionDescription('od', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg_ori = Config(od)
    cfg = get_config(cfg_ori, config_type)
    cfg.option('a').value.set(1)
    warnings.simplefilter("always", ValueWarning)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('b').value.set(1)
    assert w != []
    cfg.option('a').value.reset()
    cfg.option('b').value.set(1)
    raises(ValueError, "cfg.option('a').value.set(1)")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    warnings.simplefilter("always", ValueWarning)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a').value.set(1)
    assert len(w) == 1


def test_consistency_not_equal(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '')
    od = OptionDescription('od', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg_ori = Config(od)
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a').value.get() is None
    assert cfg.option('b').value.get() is None
    cfg.option('a').value.set(1)
    cfg.option('a').value.reset()
    cfg.option('a').value.set(1)
    raises(ValueError, "cfg.option('b').value.set(1)")
    cfg.option('b').value.set(2)
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    warnings.simplefilter("always", ValueWarning)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('b').value.set(1)
    assert len(w) == 1


def test_consistency_not_equal_many_opts(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '')
    c = IntOption('c', '')
    d = IntOption('d', '')
    e = IntOption('e', '')
    f = IntOption('f', '')
    od = OptionDescription('od', '', [a, b, c, d, e, f])
    a.impl_add_consistency('not_equal', b, c, d, e, f)
    cfg_ori = Config(od)
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a').value.get() is None
    assert cfg.option('b').value.get() is None
    #
    cfg.option('a').value.set(1)
    cfg.option('a').value.reset()
    #
    cfg.option('a').value.set(1)
    raises(ValueError, "cfg.option('b').value.set(1)")
    assert cfg.option('b').value.get() == None
    #
    cfg.option('b').value.set(2)
    raises(ValueError, "cfg.option('f').value.set(2)")
    assert cfg.option('f').value.get() is None
    assert cfg.option('a').value.get() == 1
    assert cfg.option('b').value.get() == 2
    raises(ValueError, "cfg.option('f').value.set(1)")
    assert cfg.option('f').value.get() is None
    assert cfg.option('a').value.get() == 1
    assert cfg.option('b').value.get() == 2
    #
    cfg.option('d').value.set(3)
    raises(ValueError, "cfg.option('f').value.set(3)")
    raises(ValueError, "cfg.option('a').value.set(3)")
    cfg.option('d').value.set(3)
    raises(ValueError, "cfg.option('c').value.set(3)")
    raises(ValueError, "cfg.option('e').value.set(3)")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    warnings.simplefilter("always", ValueWarning)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('c').value.set(3)
    assert len(w) == 1
    with warnings.catch_warnings(record=True) as w:
        cfg.option('e').value.set(3)
    assert len(w) == 1


def test_consistency_not_equal_many_opts_one_disabled(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '')
    c = IntOption('c', '')
    d = IntOption('d', '')
    e = IntOption('e', '')
    f = IntOption('f', '')
    g = IntOption('g', '', properties=('disabled',))
    od = OptionDescription('od', '', [a, b, c, d, e, f, g])
    a.impl_add_consistency('not_equal', b, c, d, e, f, g, transitive=False)
    cfg_ori = Config(od)
    cfg_ori.property.read_write()
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a').value.get() is None
    assert cfg.option('b').value.get() is None
    #
    cfg.option('a').value.set(1)
    cfg.option('a').value.reset()
    #
    cfg.option('a').value.set(1)
    raises(ValueError, "cfg.option('b').value.set(1)")
    #
    cfg.option('b').value.set(2)
    raises(ValueError, "cfg.option('f').value.set(2)")
    raises(ValueError, "cfg.option('f').value.set(1)")
    #
    cfg.option('d').value.set(3)
    raises(ValueError, "cfg.option('f').value.set(3)")
    raises(ValueError, "cfg.option('a').value.set(3)")
    raises(ValueError, "cfg.option('c').value.set(3)")
    raises(ValueError, "cfg.option('e').value.set(3)")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('c').value.set(3)
    assert len(w) == 1


def test_consistency_not_in_config_1():
    a = IntOption('a', '')
    b = IntOption('b', '')
    a.impl_add_consistency('not_equal', b)
    od1 = OptionDescription('od1', '', [a])
    od = OptionDescription('root', '', [od1])
    od
    raises(ConfigError, "Config(od)")


def test_consistency_not_in_config_2():
    a = IntOption('a', '')
    b = IntOption('b', '')
    a.impl_add_consistency('not_equal', b)
    od1 = OptionDescription('od1', '', [a])
    od2 = OptionDescription('od2', '', [b])
    od = OptionDescription('root', '', [od1, od2])
    Config(od)


def test_consistency_not_in_config_3():
    a = IntOption('a', '')
    b = IntOption('b', '')
    a.impl_add_consistency('not_equal', b)
    od1 = OptionDescription('od1', '', [a])
    od2 = OptionDescription('od2', '', [b])
    od = OptionDescription('root', '', [od1, od2])
    od
    #with subconfig
    raises(ConfigError, "Config(od1)")


def test_consistency_after_config():
    a = IntOption('a', '')
    b = IntOption('b', '')
    od1 = OptionDescription('od1', '', [a])
    od2 = OptionDescription('od2', '', [b])
    od = OptionDescription('root', '', [od1, od2])
    Config(od)
    raises(AttributeError, "a.impl_add_consistency('not_equal', b)")


def test_consistency_not_equal_symlink():
    a = IntOption('a', '')
    b = IntOption('b', '')
    c = SymLinkOption('c', a)
    od = OptionDescription('od', '', [a, b, c])
    a.impl_add_consistency('not_equal', b)
    cfg = Config(od)
    assert set(od._cache_consistencies.keys()) == set([a, b])


def test_consistency_mix(config_type):
    b = IntOption('b', '', multi=True)
    c = IntOption('c', '', multi=True)
    d = IntOption('d', '', multi=True)
    od = Leadership('c', '', [c, d])
    od2 = OptionDescription('a', '', [b, od])
    c.impl_add_consistency('not_equal', b, d)
    cfg_ori = Config(od2)
    cfg = get_config(cfg_ori, config_type)
    cfg.option('b').value.set([1, 2, 3])
    cfg.option('c.c').value.set([4, 5])
    raises(ValueError, "cfg.option('c.c').value.set([1, 2])")
    raises(ValueError, "cfg.option('c.d', 0).value.set(1)")
    raises(ValueError, "cfg.option('c.d', 1).value.set(4)")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('c.d', 1).value.set(4)
    assert len(w) == 1


def test_consistency_not_equal_submulti():
    a = IntOption('a', '', multi=submulti)
    b = IntOption('b', '', multi=submulti)
    od = OptionDescription('a', '', [a, b])
    raises(ConfigError, 'a.impl_add_consistency("not_equal", b)')


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(ConfigError, "a.impl_add_consistency('not_equal', b)")


def test_consistency_not_equal_leadership(config_type):
    a = IntOption('a', '', multi=True)
    b = IntOption('b', '', multi=True)
    od = Leadership('a', '', [a, b])
    od2 = OptionDescription('b', '', [od])
    a.impl_add_consistency('not_equal', b)
    cfg = Config(od2)
    cfg = get_config(cfg, config_type)
    assert cfg.option('a.a').value.get() == []
    cfg.option('a.a').value.set([1])
    cfg.option('a.a').value.reset()
    cfg.option('a.a').value.set([1])
    raises(ValueError, "cfg.option('a.b', 0).value.set(1)")
    cfg.option('a.b', 0).value.set(2)
    cfg.option('a.a').value.reset()
    cfg.option('a.a').value.set([1])
    cfg.value.dict()


def test_consistency_not_equal_leadership_error_multi1():
    a = IPOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    c = NetmaskOption('c', '', multi=True)
    od = Leadership('a', '', [a, b])
    od2 = OptionDescription('b', '', [od, c])
    c.impl_add_consistency('ip_netmask', a)
    raises(ConfigError, "Config(od2)")


def test_consistency_not_equal_leadership_error_multi2():
    a = IPOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    c = IPOption('c', '', multi=True)
    od = Leadership('a', '', [a, b])
    od2 = OptionDescription('b', '', [od, c])
    b.impl_add_consistency('ip_netmask', c)
    raises(ConfigError, "Config(od2)")


def test_consistency_ip_netmask_leadership_error_not_leader():
    a = IPOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    od = OptionDescription('a', '', [a, b])
    od2 = OptionDescription('b', '', [od])
    b.impl_add_consistency('ip_netmask', a)
    raises(ConfigError, "Config(od2)")


def test_consistency_ip_netmask_leadership_error_leader_and_not():
    a = IPOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    c = IPOption('c', '', multi=True)
    d = NetmaskOption('d', '', multi=True)
    od = Leadership('a', '', [a, b])
    od2 = OptionDescription('c', '', [c, d])
    od3 = OptionDescription('b', '', [od, od2])
    d.impl_add_consistency('ip_netmask', a)
    raises(ConfigError, "Config(od3)")


def test_consistency_ip_netmask_leadership_error_otherleader():
    a = IPOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    c = IPOption('c', '', multi=True)
    d = NetmaskOption('d', '', multi=True)
    od = Leadership('a', '', [a, b])
    od2 = Leadership('c', '', [c, d])
    od3 = OptionDescription('b', '', [od, od2])
    d.impl_add_consistency('ip_netmask', a)
    raises(ConfigError, "Config(od2)")


def test_consistency_not_equal_leadership_default():
    a = IntOption('a', '', multi=True)
    b = IntOption('b', '', multi=True, default_multi=1)
    od = Leadership('a', '', [a, b])
    od2 = OptionDescription('a', '', [od])
    a.impl_add_consistency('not_equal', b)
    cfg = Config(od2)
    # FIXME cfg = get_config(cfg, config_type)
    assert cfg.option('a.a').value.get() == []
    raises(ValueError, "cfg.option('a.a').value.set([1])")
    cfg.option('a.a').value.set([2])
    cfg.option('a.a').value.reset()
    #
    cfg.property.add('demoting_error_warning')
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a.a').value.set([1])
    assert len(w) == 1


def test_consistency_not_equal_multi(config_type):
    a = IntOption('a', '', multi=True)
    b = IntOption('b', '', multi=True)
    od = OptionDescription('a', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg_ori = Config(od)
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a').value.get() == []
    assert cfg.option('b').value.get() == []
    cfg.option('a').value.set([1])
    cfg.option('a').value.reset()
    cfg.option('a').value.set([1])
    raises(ValueError, "cfg.option('b').value.set([1])")
    cfg.option('a').value.set([2])
    raises(ValueError, "cfg.option('b').value.set([2, 1])")
    cfg.option('a').value.set([2, 3])
    raises(ValueError, "cfg.option('a').value.set([2, 3, 3])")
    raises(ValueError, "cfg.option('b').value.set([2, 3])")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('b').value.set([2, 3])
    assert len(w) == 1


def test_consistency_not_equal_multi_default1():
    a = IntOption('a', '', multi=True, default=[1])
    b = IntOption('b', '', multi=True, default=[3, 1])
    od = OptionDescription('a', '', [a, b])
    # FIXME raises(ValueError, "b.impl_add_consistency('not_equal', a)")


def test_consistency_not_equal_multi_default2():
    a = IntOption('a', '', multi=True, default=[1])
    b = IntOption('b', '', multi=True, default_multi=1)
    od = OptionDescription('a', '', [a, b])
    #default_multi not tested now
    a.impl_add_consistency('not_equal', b)


def test_consistency_not_equal_leader_default(config_type):
    a = IntOption('a', '', multi=True, default=[2, 1])
    b = IntOption('b', '', multi=True, default_multi=1)
    od = Leadership('a', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    od2 = OptionDescription('a', '', [od])
    cfg = Config(od2)
    cfg = get_config(cfg, config_type)
    # default_multi not tested
    raises(ValueError, "cfg.option('a.b', 0).value.get()")
    cfg.option('a.b', 0).value.set(3)
    cfg.option('a.b', 1).value.set(3)
    assert cfg.option('a.b', 1).value.get() == 3


def test_consistency_not_equal_multi_default_modif(config_type):
    a = IntOption('a', '', multi=True)
    b = IntOption('b', '', multi=True, default=[1, 2])
    od = OptionDescription('a', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg_ori = Config(od)
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a').value.get() == []
    assert cfg.option('b').value.get() == [1, 2]
    raises(ValueError, "cfg.option('a').value.set([1])")
    raises(ValueError, "cfg.option('b').value.set([1, 2, 1])")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('b').value.set([1, 2, 1])
    assert len(w) == 1


def test_consistency_default():
    a = IntOption('a', '', 1)
    b = IntOption('b', '', 1)
    a, b
    raises(ValueError, "a.impl_add_consistency('not_equal', b)")


def test_consistency_default_multi():
    a = IntOption('a', '', [2, 1], multi=True)
    b = IntOption('b', '', [1, 1], multi=True)
    c = IntOption('c', '', [1, 2], multi=True)
    b
    # FIXME raises(ValueError, "a.impl_add_consistency('not_equal', b)")
    # FIXME raises(ValueError, "a.impl_add_consistency('not_equal', c)")


def test_consistency_default_diff():
    a = IntOption('a', '', 3)
    b = IntOption('b', '', 1)
    od = OptionDescription('od', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg = Config(od)
    # FIXME cfg = get_config(cfg, config_type)
    raises(ValueError, "cfg.option('a').value.set(1)")
    cfg.option('a').value.set(2)
    cfg.option('b').value.set(3)
    owner = cfg.owner.get()
    assert cfg.option('a').owner.get() == owner
    raises(ValueError, "cfg.option('a').value.reset()")
    assert cfg.option('a').owner.get() == owner
    #
    cfg.property.add('demoting_error_warning')
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a').value.reset()
    assert len(w) == 1


def test_consistency_ip_netmask(config_type):
    a = IPOption('a', '')
    b = NetmaskOption('b', '')
    od = OptionDescription('od', '', [a, b])
    b.impl_add_consistency('ip_netmask', a)
    cfg_ori = Config(od)
    cfg = cfg_ori
    cfg = get_config(cfg_ori, config_type)
    cfg.option('a').value.set('192.168.1.1')
    cfg.option('b').value.set('255.255.255.0')
    cfg.option('a').value.set('192.168.1.2')
    cfg.option('b').value.set('255.255.255.128')
    cfg.option('b').value.set('255.255.255.0')
    raises(ValueError, "cfg.option('a').value.set('192.168.1.0')")
    raises(ValueError, "cfg.option('a').value.set('192.168.1.255')")
    cfg.option('a').value.reset()
    cfg.option('b').value.reset()
    cfg.option('a').value.set('192.168.1.255')
    raises(ValueError, "cfg.option('b').value.set('255.255.255.0')")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('b').value.set('255.255.255.0')
    assert len(w) == 1


def test_consistency_ip_netmask_invalid():
    b = NetmaskOption('b', '')
    od = OptionDescription('od', '', [b])
    raises(ConfigError, "b.impl_add_consistency('ip_netmask')")


def test_consistency_network_netmask(config_type):
    a = NetworkOption('a', '')
    b = NetmaskOption('b', '')
    od = OptionDescription('od', '', [a, b])
    b.impl_add_consistency('network_netmask', a)
    cfg_ori = Config(od)
    cfg = get_config(cfg_ori, config_type)
    cfg.option('a').value.set('192.168.1.1')
    cfg.option('b').value.set('255.255.255.255')
    cfg.option('b').value.reset()
    cfg.option('a').value.set('192.168.1.0')
    cfg.option('b').value.set('255.255.255.0')
    raises(ValueError, "cfg.option('a').value.set('192.168.1.1')")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a').value.set('192.168.1.1')
    assert len(w) == 1


def test_consistency_network_netmask_invalid():
    b = NetmaskOption('b', '')
    od = OptionDescription('od', '', [b])
    raises(ConfigError, "b.impl_add_consistency('network_netmask')")


def test_consistency_ip_in_network(config_type):
    a = NetworkOption('a', '')
    b = NetmaskOption('b', '')
    c = IPOption('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 = get_config(cfg, config_type)
    cfg.option('a').value.set('192.168.1.0')
    cfg.option('b').value.set('255.255.255.0')
    cfg.option('c').value.set('192.168.1.1')
    raises(ValueError, "cfg.option('c').value.set('192.168.2.1')")
    raises(ValueError, "cfg.option('c').value.set('192.168.1.0')")
    raises(ValueError, "cfg.option('c').value.set('192.168.1.255')")
    with warnings.catch_warnings(record=True) as w:
        cfg.option('d').value.set('192.168.2.1')
    assert len(w) == 1


def test_consistency_ip_in_network_cidr(config_type):
    a = NetworkOption('a', '', cidr=True)
    c = IPOption('c', '')
    d = IPOption('d', '')
    od = OptionDescription('od', '', [a, c, d])
    c.impl_add_consistency('in_network', a)
    d.impl_add_consistency('in_network', a, warnings_only=True)
    warnings.simplefilter("always", ValueWarning)
    cfg = Config(od)
    cfg = get_config(cfg, config_type)
    cfg.option('a').value.set('192.168.1.0/24')
    cfg.option('c').value.set('192.168.1.1')
    raises(ValueError, "cfg.option('c').value.set('192.168.2.1')")
    raises(ValueError, "cfg.option('c').value.set('192.168.1.0')")
    raises(ValueError, "cfg.option('c').value.set('192.168.1.255')")
    with warnings.catch_warnings(record=True) as w:
        cfg.option('d').value.set('192.168.2.1')
    assert len(w) == 1


def test_consistency_ip_in_network_invalid():
    a = NetworkOption('a', '')
    b = NetmaskOption('b', '')
    c = IPOption('c', '')
    d = IPOption('d', '')
    od = OptionDescription('od', '', [a, b, c, d])
    raises(ConfigError, "c.impl_add_consistency('in_network', a)")


def test_consistency_ip_netmask_error_multi():
    a = IPOption('a', '', multi=True)
    b = NetmaskOption('b', '')
    OptionDescription('od', '', [a, b])
    raises(ConfigError, "b.impl_add_consistency('ip_netmask', a)")


def test_consistency_ip_netmask_multi(config_type):
    a = IPOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    od = Leadership('a', '', [a, b])
    b.impl_add_consistency('ip_netmask', a)
    od2 = OptionDescription('od2', '', [od])
    cfg_ori = Config(od2)
    cfg = get_config(cfg_ori, config_type)
    cfg.option('a.a').value.set(['192.168.1.1'])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    cfg.option('a.a').value.set(['192.168.1.2'])
    cfg.option('a.b', 0).value.set('255.255.255.128')
    cfg.option('a.b', 0).value.set('255.255.255.0')
    raises(ValueError, "cfg.option('a.a').value.set(['192.168.1.0'])")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a.a').value.set(['192.168.1.0'])
    assert len(w) == 1


def test_consistency_network_netmask_multi(config_type):
    a = NetworkOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    od = Leadership('a', '', [a, b])
    b.impl_add_consistency('network_netmask', a)
    od2 = OptionDescription('od', '', [od])
    cfg = Config(od2)
    cfg = get_config(cfg, config_type)
    cfg.option('a.a').value.set(['192.168.1.1'])
    cfg.option('a.b', 0).value.set('255.255.255.255')
    cfg.option('a.b', 0).value.reset()
    cfg.option('a.a').value.set(['192.168.1.0'])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    raises(ValueError, "cfg.option('a.a').value.set(['192.168.1.1'])")


def test_consistency_network_netmask_multi_follower_default_multi(config_type):
    a = NetworkOption('a', '', default_multi=u'192.168.1.0', multi=True, properties=('mandatory',))
    b = NetmaskOption('b', '', default_multi=u'255.255.255.0', multi=True, properties=('mandatory',))
    od = Leadership('a', '', [a, b])
    od2 = OptionDescription('od2', '', [od])
    b.impl_add_consistency('network_netmask', a)
    cfg = Config(od2)
    cfg.property.read_write()
    cfg = get_config(cfg, config_type)
    cfg.option('a.a').value.set([undefined])
    assert cfg.option('a.a').value.get() == ['192.168.1.0']
    assert cfg.option('a.b', 0).value.get() == '255.255.255.0'


def test_consistency_network_netmask_multi_follower_default(config_type):
    a = NetworkOption('a', '', multi=True, properties=('mandatory',))
    b = NetmaskOption('b', '', default_multi=u'255.255.255.0', multi=True, properties=('mandatory',))
    od = Leadership('a', '', [a, b])
    b.impl_add_consistency('network_netmask', a)
    od2 = OptionDescription('od2', '', [od])
    cfg_ori = Config(od2)
    cfg_ori.property.read_write()
    cfg_ori.property.pop('cache')
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a.a').value.get() == []
    cfg.option('a.a').value.set(['192.168.1.0'])
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.read_only()
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a.a').value.get() == [u'192.168.1.0']
    assert cfg.option('a.b', 0).value.get() == u'255.255.255.0'
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.read_write()
    cfg = get_config(cfg_ori, config_type)
    raises(ValueError, "cfg.option('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])")
    cfg.option('a.a').value.set(['192.168.1.0', undefined])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    cfg.option('a.b', 1).value.set('255.255.255.255')
    cfg.option('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])


def return_netmask(*args, **kwargs):
    return u'255.255.255.0'


def return_netmask2(leader):
    if leader is not None:
        if leader.endswith('2.1'):
            return u'255.255.255.0'
        if not leader.endswith('.0'):
            return u'255.255.255.255'
    return u'255.255.255.0'


def test_consistency_network_netmask_multi_follower_callback(config_type):
    a = NetworkOption('a', '', multi=True, properties=('mandatory',))
    b = NetmaskOption('b', '', Calculation(return_netmask, Params(kwargs={'index': ParamIndex()})), multi=True, properties=('mandatory',))
    od = Leadership('a', '', [a, b])
    b.impl_add_consistency('network_netmask', a)
    od2 = OptionDescription('od2', '', [od])
    cfg_ori = Config(od2)
    cfg_ori.property.read_write()
    cfg_ori.property.pop('cache')
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a.a').value.get() == []
    cfg.option('a.a').value.set(['192.168.1.0'])
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.read_only()
    cfg = get_config(cfg_ori, config_type)
    assert cfg.option('a.a').value.get() == [u'192.168.1.0']
    assert cfg.option('a.b', 0).value.get() == '255.255.255.0'
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.read_write()
    cfg = get_config(cfg_ori, config_type)
    raises(ValueError, "assert cfg.option('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])")
    cfg.option('a.a').value.set(['192.168.1.0', undefined])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    cfg.option('a.b', 1).value.set('255.255.255.255')
    cfg.option('a.a').value.set(['192.168.1.0', '192.168.1.1'])


def test_consistency_network_netmask_multi_follower_callback_value(config_type):
    a = NetworkOption('a', '', multi=True, properties=('mandatory',))
    b = NetmaskOption('b', '', Calculation(return_netmask2, Params(ParamOption(a))), multi=True, properties=('mandatory',))
    od = Leadership('a', '', [a, b])
    b.impl_add_consistency('network_netmask', a)
    od2 = OptionDescription('od2', '', [od])
    cfg = Config(od2)
    cfg.property.read_write()
    cfg.property.pop('cache')
    cfg = get_config(cfg, config_type)
    assert cfg.option('a.a').value.get() == []
    cfg.option('a.a').value.set(['192.168.1.0'])
    assert cfg.option('a.a').value.get() == ['192.168.1.0']
    assert cfg.option('a.b', 0).value.get() == '255.255.255.0'
    raises(ValueError, "cfg.option('a.a').value.set(['192.168.1.0', '192.168.2.1'])")
    assert cfg.option('a.a').value.get() == [u'192.168.1.0']
    assert cfg.option('a.b', 0).value.get() == '255.255.255.0'
    raises(ValueError, "cfg.option('a.a').value.set(['192.168.2.1'])")
    assert cfg.option('a.a').value.get() == [u'192.168.1.0']
    assert cfg.option('a.b', 0).value.get() == '255.255.255.0'
    cfg.option('a.a').value.set(['192.168.1.0', '192.168.1.1'])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    cfg.option('a.b', 1).value.set('255.255.255.255')


def test_consistency_ip_netmask_multi_leader(config_type):
    a = IPOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    od = Leadership('a', '', [a, b])
    b.impl_add_consistency('ip_netmask', a)
    od2 = OptionDescription('od2', '', [od])
    cfg = Config(od2)
    cfg = get_config(cfg, config_type)
    cfg.option('a.a').value.set(['192.168.1.1'])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    cfg.option('a.a').value.set(['192.168.1.2'])
    cfg.option('a.b', 0).value.set('255.255.255.128')
    cfg.option('a.b', 0).value.set('255.255.255.0')
    raises(ValueError, "cfg.option('a.a').value.set(['192.168.1.0'])")
    cfg.option('a.a').value.set(['192.168.1.128'])
    raises(ValueError, "cfg.option('a.b', 0).value.set('255.255.255.128')")
    cfg.option('a.a').value.set(['192.168.1.2', '192.168.1.3'])


def test_consistency_network_netmask_multi_leader(config_type):
    a = NetworkOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    od = Leadership('a', '', [a, b])
    b.impl_add_consistency('network_netmask', a)
    od2 = OptionDescription('od2', '', [od])
    cfg = Config(od2)
    cfg = get_config(cfg, config_type)
    cfg.option('a.a').value.set(['192.168.1.1'])
    cfg.option('a.b', 0).value.set('255.255.255.255')
    cfg.option('a.b', 0).value.reset()
    cfg.option('a.a').value.set(['192.168.1.0'])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    raises(ValueError, "cfg.option('a.a').value.set(['192.168.1.1'])")


def test_consistency_broadcast(config_type):
    a = NetworkOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    c = BroadcastOption('c', '', multi=True)
    od = Leadership('a', '', [a, b, c])
    b.impl_add_consistency('network_netmask', a)
    c.impl_add_consistency('broadcast', a, b)
    od2 = OptionDescription('od2', '', [od])
    cfg = Config(od2)
    cfg = get_config(cfg, config_type)
    #first, test network_netmask
    cfg.option('a.a').value.set(['192.168.1.128'])
    raises(ValueError, "cfg.option('a.a').value.set(['255.255.255.0'])")
    #
    cfg.option('a.a').value.set(['192.168.1.0'])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    cfg.option('a.c', 0).value.set('192.168.1.255')
    raises(ValueError, "cfg.option('a.a').value.set(['192.168.1.1'])")
    #
    cfg.option('a.a').value.set(['192.168.1.0', '192.168.2.128'])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    cfg.option('a.b', 1).value.set('255.255.255.128')
    cfg.option('a.c', 0).value.set('192.168.1.255')
    cfg.option('a.c', 1).value.set('192.168.2.255')
    raises(ValueError, "cfg.option('a.c', 1).value.set('192.168.2.128')")
    cfg.option('a.c', 1).value.set('192.168.2.255')


def test_consistency_broadcast_error(config_type):
    a = NetworkOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    c = BroadcastOption('c', '', multi=True)
    od = Leadership('a', '', [a, b, c])
    od2 = OptionDescription('od2', '', [od])
    b.impl_add_consistency('network_netmask', a)
    c.impl_add_consistency('broadcast', a)
    cfg = Config(od2)
    cfg = get_config(cfg, config_type)
    raises(ConfigError, "cfg.option('a.a').value.set(['192.168.1.0'])")


def test_consistency_broadcast_warnings(config_type):
    warnings.simplefilter("always", ValueWarning)
    a = NetworkOption('a', '', properties=('mandatory', 'disabled'))
    b = NetmaskOption('b', '', properties=('mandatory', 'disabled'))
    c = NetmaskOption('c', '', properties=('mandatory', 'disabled'))
    od = OptionDescription('a', '', [a, b, c])
    b.impl_add_consistency('network_netmask', a, warnings_only=True)
    cfg_ori = Config(od)
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a').value.set('192.168.1.4')
        cfg.option('b').value.set('255.255.255.0')
    assert len(w) == 1
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.read_write()
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        list(cfg.value.mandatory())
    assert len(w) == 0


def test_consistency_broadcast_default_1():
    a = NetworkOption('a', '', '192.168.1.0')
    b = NetmaskOption('b', '', '255.255.255.128')
    c = BroadcastOption('c', '', '192.168.2.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')
    d = BroadcastOption('d', '', '192.168.1.127')
    od2 = OptionDescription('a', '', [a, b, d])
    od2
    d.impl_add_consistency('broadcast', a, b)


def test_consistency_not_all(config_type):
    #_cache_consistencies is not None by not options has consistencies
    a = NetworkOption('a', '', multi=True)
    b = NetmaskOption('b', '', multi=True)
    c = BroadcastOption('c', '', multi=True)
    od = Leadership('a', '', [a, b, c])
    b.impl_add_consistency('network_netmask', a)
    od2 = OptionDescription('od2', '', [od])
    cfg = Config(od2)
    cfg = get_config(cfg, config_type)
    cfg.option('a.a').value.set(['192.168.1.0'])
    cfg.option('a.b', 0).value.set('255.255.255.0')
    cfg.option('a.c', 0).value.set('192.168.1.255')


def test_consistency_permissive(config_type):
    a = IntOption('a', '', 1)
    b = IntOption('b', '', 2, properties=('hidden',))
    od = OptionDescription('od', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg = Config(od)
    cfg.property.read_write()
    cfg.permissive.set(frozenset(['hidden']))
    cfg = get_config(cfg, config_type)
    cfg.option('a').value.set(1)


def test_consistency_disabled(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '', properties=('disabled',))
    od = OptionDescription('od', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg = Config(od)
    cfg.property.read_write()
    cfg = get_config(cfg, config_type)
    raises(PropertiesOptionError, "cfg.option('a').value.set(1)")


def test_consistency_disabled_transitive(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '', properties=('disabled',))
    od = OptionDescription('od', '', [a, b])
    a.impl_add_consistency('not_equal', b, transitive=False)
    cfg = Config(od)
    cfg.property.read_write()
    cfg = get_config(cfg, config_type)
    cfg.option('a').value.set(1)


def test_consistency_disabled_transitive_2(config_type):
    a = IPOption('a', '')
    b = IPOption('b', '')
    c = NetworkOption('c', '', default='192.168.1.0')
    d = NetmaskOption('d', '', default='255.255.255.0', properties=('disabled',))
    od = OptionDescription('od', '', [a, b, c, d])
    a.impl_add_consistency('not_equal', b)
    a.impl_add_consistency('in_network', c, d, transitive=False)
    cfg_ori = Config(od)
    cfg_ori.property.read_write()
    cfg = get_config(cfg_ori, config_type)
    cfg.option('a').value.set('192.168.1.1')
    raises(ValueError, "cfg.option('b').value.set('192.168.1.1')")
    cfg.option('a').value.set('192.168.2.1')
    #
    cfg.option('a').value.set('192.168.1.1')
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.pop('disabled')
    cfg = get_config(cfg_ori, config_type)
    raises(ValueError, "cfg.option('a').value.set('192.168.2.1')")
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.add('demoting_error_warning')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a').value.set('192.168.2.1')
    assert len(w) == 1


def return_val(*args, **kwargs):
    return '192.168.1.1'


def test_consistency_with_callback(config_type):
    a = NetworkOption('a', '', default='192.168.1.0')
    b = NetmaskOption('b', '', default='255.255.255.0')
    c = IPOption('c', '', Calculation(return_val, Params(ParamOption(a))))
    od = OptionDescription('od', '', [a, b, c])
    c.impl_add_consistency('in_network', a, b)
    cfg = Config(od)
    cfg = get_config(cfg, config_type)
    cfg.option('c').value.get()


def test_consistency_warnings_only_options(config_type):
    a = IPOption('a', '', warnings_only=True)
    b = IPOption('b', '')
    c = NetworkOption('c', '', default='192.168.1.0')
    d = NetmaskOption('d', '', default='255.255.255.0', properties=('disabled',))
    od = OptionDescription('od', '', [a, b, c, d])
    a.impl_add_consistency('not_equal', b)
    a.impl_add_consistency('in_network', c, d, transitive=False)
    cfg_ori = Config(od)
    cfg_ori.property.read_write()
    cfg = get_config(cfg_ori, config_type)
    cfg.option('a').value.set('192.168.1.1')
    raises(ValueError, "cfg.option('b').value.set('192.168.1.1')")
    cfg.option('a').value.set('192.168.2.1')
    #
    cfg.option('a').value.set('192.168.1.1')
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.pop('disabled')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a').value.set('192.168.2.1')
    assert len(w) == 1


def test_consistency_warnings_only_options_callback(config_type):
    a = IPOption('a', '', warnings_only=True)
    b = IPOption('b', '')
    c = NetworkOption('c', '', default='192.168.1.0')
    d = NetmaskOption('d', '', Calculation(return_netmask2, Params(ParamOption(c))))
    od = OptionDescription('od', '', [a, b, c, d])
    a.impl_add_consistency('not_equal', b)
    a.impl_add_consistency('in_network', c, d, transitive=False)
    cfg_ori = Config(od)
    cfg_ori.property.read_write()
    cfg = get_config(cfg_ori, config_type)
    cfg.option('a').value.set('192.168.1.1')
    raises(ValueError, "cfg.option('b').value.set('192.168.1.1')")
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a').value.set('192.168.2.1')
    assert len(w) == 1
    #
    cfg.option('a').value.set('192.168.1.1')
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.pop('disabled')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('a').value.set('192.168.2.1')
    assert len(w) == 1


def test_consistency_double_warnings(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '', 1)
    c = IntOption('c', '', 1)
    od = OptionDescription('od', '', [a, b, c])
    warnings.simplefilter("always", ValueWarning)
    a.impl_add_consistency('not_equal', b, warnings_only=True)
    a.impl_add_consistency('not_equal', c, warnings_only=True)
    od2 = OptionDescription('od2', '', [od])
    cfg_ori = Config(od2)
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('od.a').value.set(1)
    assert w != []
    if config_type == 'tiramisu-api':
        # in this case warnings is for '"a" and "b"'
        assert len(w) == 1
    else:
        # in this cas one warnings is for "a" and the second for "b"
        assert len(w) == 2
    with warnings.catch_warnings(record=True) as w:
        cfg.option('od.c').value.set(2)
    assert len(w) == 0
    with warnings.catch_warnings(record=True) as w:
        cfg.option('od.a').value.set(2)
    assert len(w) == 1
    #
    if config_type == 'tiramisu-api':
        cfg.send()
    cfg_ori.property.pop('warnings')
    cfg = get_config(cfg_ori, config_type)
    with warnings.catch_warnings(record=True) as w:
        cfg.option('od.a').value.set(1)
    assert w == []


def test_consistency_warnings_error(config_type):
    a = IntOption('a', '')
    b = IntOption('b', '', 1)
    c = IntOption('c', '', 1)
    od = OptionDescription('od', '', [a, b, c])
    warnings.simplefilter("always", ValueWarning)
    a.impl_add_consistency('not_equal', b, warnings_only=True)
    a.impl_add_consistency('not_equal', c)
    cfg = Config(od)
    cfg = get_config(cfg, config_type)
    with warnings.catch_warnings(record=True) as w:
        raises(ValueError, "cfg.option('a').value.set(1)")
    assert w == []


def test_consistency_network_netmask_mandatory(config_type):
    a = NetworkOption('a', '', multi=True, properties=('mandatory',), default=[u'0.0.0.0'])
    b = NetmaskOption('b', '', multi=True, properties=('mandatory',), default_multi=u'0.0.0.0')
    od = Leadership('a', '', [a, b])
    b.impl_add_consistency('network_netmask', a)
    od2 = OptionDescription('od2', '', [od])
    cfg = Config(od2)
    cfg.property.read_only()
    cfg.property.pop('mandatory')
    cfg = get_config(cfg, config_type)
    cfg.value.dict()


def test_consistency_has_dependency():
    a = IPOption('a', '')
    b = NetmaskOption('b', '')
    od = OptionDescription('od', '', [a, b])
    b.impl_add_consistency('ip_netmask', a)
    cfg = Config(od)
    assert cfg.option('a').option.has_dependency() is True
    assert cfg.option('b').option.has_dependency() is True
    assert cfg.option('a').option.has_dependency(False) is True
    assert cfg.option('b').option.has_dependency(False) is True


def test_consistency_not_equal_has_dependency():
    a = IntOption('a', '')
    b = IntOption('b', '')
    od = OptionDescription('od', '', [a, b])
    a.impl_add_consistency('not_equal', b)
    cfg = Config(od)
    assert cfg.option('a').option.has_dependency() is False
    assert cfg.option('b').option.has_dependency() is False
    assert cfg.option('a').option.has_dependency(False) is True
    assert cfg.option('b').option.has_dependency(False) is True