tiramisu/tests/test_symlink.py

# coding: utf-8
import pytest
from .autopath import do_autopath
do_autopath()
from .config import config_type, get_config

from tiramisu import BoolOption, StrOption, SymLinkOption, \
    OptionDescription, Leadership, Config, Calculation, calc_value, Params, ParamOption, ParamValue
from tiramisu.error import PropertiesOptionError, ConfigError
from tiramisu.setting import groups, owners
from tiramisu.storage import list_sessions


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


def return_value():
    pass


#____________________________________________________________
@pytest.mark.asyncio
async def test_symlink_option(config_type):
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    cfg = await get_config(cfg, config_type)
    assert await cfg.option('s1.b').value.get() is False
    await cfg.option("s1.b").value.set(True)
    await cfg.option("s1.b").value.set(False)
    assert await cfg.option('s1.b').value.get() is False
    assert await cfg.option('c').value.get() is False
    await cfg.option('s1.b').value.set(True)
    assert await cfg.option('s1.b').value.get() is True
    assert await cfg.option('c').value.get() is True
    await cfg.option('s1.b').value.set(False)
    assert await cfg.option('s1.b').value.get() is False
    assert await cfg.option('c').value.get() is False


@pytest.mark.asyncio
async def test_symlink_assign_option(config_type):
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    cfg = await get_config(cfg, config_type)
    with pytest.raises(ConfigError): 
        await cfg.option('c').value.set(True)


@pytest.mark.asyncio
async def test_symlink_del_option(config_type):
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    cfg = await get_config(cfg, config_type)
    with pytest.raises(ConfigError): 
        await cfg.option('c').value.reset()


@pytest.mark.asyncio
async def test_symlink_addproperties():
    boolopt = BoolOption('b', '', default=True, properties=('test',))
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription('opt', '', [boolopt, linkopt])
    cfg = await Config(descr)
    await cfg.property.read_write()
    with pytest.raises(TypeError): 
        await cfg.option('c').property.add('new')
    try:
        await cfg.option('c').property.reset()
    except AssertionError:
        pass
    else:
        raise Exception('must raise')


@pytest.mark.asyncio
async def test_symlink_getpermissive():
    boolopt = BoolOption('b', '', default=True, properties=('test',))
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription('opt', '', [boolopt, linkopt])
    cfg = await Config(descr)
    await cfg.property.read_write()
    await cfg.option('b').permissive.set(frozenset(['perm']))
    await cfg.option('c').permissive.get() == frozenset(['perm'])


@pytest.mark.asyncio
async def test_symlink_addpermissives():
    boolopt = BoolOption('b', '', default=True, properties=('test',))
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription('opt', '', [boolopt, linkopt])
    cfg = await Config(descr)
    await cfg.property.read_write()
    with pytest.raises(TypeError): 
        await cfg.option('c').permissive.set(frozenset(['new']))
    try:
        await cfg.option('c').permissive.reset()
    except AssertionError:
        pass
    else:
        raise Exception('must raise')


@pytest.mark.asyncio
async def test_symlink_getproperties():
    boolopt = BoolOption('b', '', default=True, properties=('test',))
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription('opt', '', [boolopt, linkopt])
    cfg = await Config(descr)
    await cfg.property.read_write()
    assert boolopt.impl_getproperties() == linkopt.impl_getproperties() == {'test'}
    assert boolopt.impl_has_callback() == linkopt.impl_has_callback() == False


@pytest.mark.asyncio
async def test_symlink_getcallback():
    boolopt = BoolOption('b', '', Calculation(return_value))
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription('opt', '', [boolopt, linkopt])
    cfg = await Config(descr)
    await cfg.property.read_write()
    #assert boolopt.impl_has_callback() == linkopt.impl_has_callback() == True
    #assert boolopt.impl_get_callback() == linkopt.impl_get_callback() == (return_value, None)
    assert boolopt.impl_has_callback() == linkopt.impl_has_callback() == False


@pytest.mark.asyncio
async def test_symlink_requires(config_type):
    boolopt = BoolOption('b', '', default=True)
    disabled_property = Calculation(calc_value,
                                    Params(ParamValue('disabled'),
                                           kwargs={'condition': ParamOption(boolopt),
                                                   'expected': ParamValue(False)}))
    stropt = StrOption('s', '', properties=(disabled_property,))
    linkopt = SymLinkOption("c", stropt)
    descr = OptionDescription('opt', '', [boolopt, stropt, linkopt])
    cfg = await Config(descr)
    await cfg.property.read_write()
    cfg = await get_config(cfg, config_type)
    assert await cfg.option('b').value.get() is True
    assert await cfg.option('s').value.get() is None
    assert await cfg.option('c').value.get() is None
    await cfg.option('b').value.set(False)
    #
    props = []
    try:
        await cfg.option('s').value.get()
    except PropertiesOptionError as err:
        props = err.proptype
    assert props == {'disabled'}
    #
    props = []
    try:
        await cfg.option('c').value.get()
    except PropertiesOptionError as err:
        props = err.proptype
    assert props == {'disabled'}


@pytest.mark.asyncio
async def test_symlink_multi(config_type):
    boolopt = BoolOption("b", "", default=[False], multi=True)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    cfg = await get_config(cfg, config_type)
    assert await cfg.option('s1.b').value.get() == [False]
    assert await cfg.option('c').value.get() == [False]
    await cfg.option('s1.b').value.set([True])
    assert await cfg.option('s1.b').value.get() == [True]
    assert await cfg.option('c').value.get() == [True]
    await cfg.option('s1.b').value.set([False])
    assert await cfg.option('s1.b').value.get() == [False]
    assert await cfg.option('c').value.get() == [False]
    await cfg.option('s1.b').value.set([False, True])
    assert await cfg.option('s1.b').value.get() == [False, True]
    assert await cfg.option('c').value.get() == [False, True]
    assert boolopt.impl_is_multi() is True
    assert linkopt.impl_is_multi() is True


@pytest.mark.asyncio
async def test_symlink_assign(config_type):
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    cfg = await get_config(cfg, config_type)
    with pytest.raises(ConfigError): 
        await cfg.option('c').value.set(True)


@pytest.mark.asyncio
async def test_symlink_owner(config_type):
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    cfg = await get_config(cfg, config_type)
    assert await cfg.option('s1.b').owner.isdefault()
    assert await cfg.option('c').owner.isdefault()
    await cfg.option('s1.b').value.set(True)
    assert not await cfg.option('s1.b').owner.isdefault()
    assert not await cfg.option('c').owner.isdefault()


@pytest.mark.asyncio
async def test_symlink_get_information():
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    boolopt.impl_set_information('test', 'test')
    assert boolopt.impl_get_information('test') == 'test'
    assert linkopt.impl_get_information('test') == 'test'
    boolopt.impl_set_information('test', 'test2')
    assert boolopt.impl_get_information('test') == 'test2'
    assert linkopt.impl_get_information('test') == 'test2'


@pytest.mark.asyncio
async def test_symlink_leader():
    a = StrOption('a', "", multi=True)
    ip_admin_eth0 = SymLinkOption('ip_admin_eth0', a)
    netmask_admin_eth0 = StrOption('netmask_admin_eth0', "", multi=True)
    with pytest.raises(ValueError): 
        Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])


@pytest.mark.asyncio
async def test_symlink_followers():
    a = StrOption('a', "", multi=True)
    ip_admin_eth0 = StrOption('ip_admin_eth0', "ip réseau autorisé", multi=True)
    netmask_admin_eth0 = SymLinkOption('netmask_admin_eth0', a)
    with pytest.raises(ValueError): 
        Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])


@pytest.mark.asyncio
async def test_symlink_with_leader(config_type):
    ip_admin_eth0 = StrOption('ip_admin_eth0', "ip réseau autorisé", multi=True)
    netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-réseau", multi=True)
    interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
    leader = SymLinkOption('leader', ip_admin_eth0)
    od = OptionDescription('root', '', [interface1, leader])
    cfg = await Config(od)
    cfg = await get_config(cfg, config_type)
    assert await cfg.value.dict() == {'ip_admin_eth0.ip_admin_eth0': [], 'ip_admin_eth0.netmask_admin_eth0': [], 'leader': []}
    await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val2'])
    assert await cfg.value.dict() == {'ip_admin_eth0.ip_admin_eth0': ['val1', 'val2'], 'ip_admin_eth0.netmask_admin_eth0': [None, None], 'leader': ['val1', 'val2']}


@pytest.mark.asyncio
async def test_symlink_with_follower(config_type):
    ip_admin_eth0 = StrOption('ip_admin_eth0', "ip réseau autorisé", multi=True)
    netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-réseau", multi=True)
    interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
    follower = SymLinkOption('follower', netmask_admin_eth0)
    od = OptionDescription('root', '', [interface1, follower])
    cfg = await Config(od)
    cfg = await get_config(cfg, config_type)
    assert await cfg.value.dict() == {'ip_admin_eth0.ip_admin_eth0': [], 'ip_admin_eth0.netmask_admin_eth0': [], 'follower': []}
    await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val2'])
    assert await cfg.value.dict() == {'ip_admin_eth0.ip_admin_eth0': ['val1', 'val2'], 'ip_admin_eth0.netmask_admin_eth0': [None, None], 'follower': [None, None]}
    #
    assert await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.get() == None
    assert await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.get() == None
    assert await cfg.option('follower', 0).value.get() == None
    assert await cfg.option('follower', 1).value.get() == None
    #
    await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val3')
    assert await cfg.value.dict() == {'ip_admin_eth0.ip_admin_eth0': ['val1', 'val2'], 'ip_admin_eth0.netmask_admin_eth0': [None, 'val3'], 'follower': [None, 'val3']}
    #
    assert await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.get() == None
    assert await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.get() == 'val3'
    assert await cfg.option('follower', 0).value.get() == None
    assert await cfg.option('follower', 1).value.get() == 'val3'


#____________________________________________________________
@pytest.mark.asyncio
async def test_symlink_dependency():
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    assert await cfg.option('s1.b').option.has_dependency() is False
    assert await cfg.option('c').option.has_dependency() is True
    assert await cfg.option('s1.b').option.has_dependency(False) is True
    assert await cfg.option('c').option.has_dependency(False) is False


@pytest.mark.asyncio
async def test_symlink_makedict(config_type):
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    cfg = await get_config(cfg, config_type)
    assert await cfg.value.dict() == {'c': False, 's1.b': False}
    await cfg.option('s1.b').value.set(True)
    assert await cfg.value.dict() == {'c': True, 's1.b': True}


@pytest.mark.asyncio
async def test_symlink_list(config_type):
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    cfg = await Config(descr)
    cfg = await get_config(cfg, config_type)
    list_opt = []
    for opt in await cfg.option.list():
        list_opt.append(await opt.option.path())
    assert list_opt == ['c']
    #
    list_opt = []
    for opt in await cfg.option.list(recursive=True):
        list_opt.append(await opt.option.path())
    assert list_opt == ['c', 's1.b']