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
from .config import event_loop


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])])
    async with await Config(descr) as cfg:
        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
    assert not await list_sessions()


@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])])
    async with await Config(descr) as cfg:
        cfg = await get_config(cfg, config_type)
        with pytest.raises(ConfigError):
            await cfg.option('c').value.set(True)
    assert not await list_sessions()


@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])])
    async with await Config(descr) as cfg:
        cfg = await get_config(cfg, config_type)
        with pytest.raises(ConfigError):
            await cfg.option('c').value.reset()
    assert not await list_sessions()


@pytest.mark.asyncio
async def test_symlink_addproperties():
    boolopt = BoolOption('b', '', default=True, properties=('test',))
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription('opt', '', [boolopt, linkopt])
    async with await Config(descr) as cfg:
        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')
    assert not await list_sessions()


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


@pytest.mark.asyncio
async def test_symlink_addpermissives():
    boolopt = BoolOption('b', '', default=True, properties=('test',))
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription('opt', '', [boolopt, linkopt])
    async with await Config(descr) as cfg:
        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')
    assert not await list_sessions()


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


@pytest.mark.asyncio
async def test_symlink_getcallback():
    boolopt = BoolOption('b', '', Calculation(return_value))
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription('opt', '', [boolopt, linkopt])
    async with await Config(descr) as cfg:
        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
    assert not await list_sessions()


@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])
    async with await Config(descr) as cfg:
        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'}
    assert not await list_sessions()


@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])])
    async with await Config(descr) as cfg:
        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
    assert not await list_sessions()


@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])])
    async with await Config(descr) as cfg:
        cfg = await get_config(cfg, config_type)
        with pytest.raises(ConfigError):
            await cfg.option('c').value.set(True)
    assert not await list_sessions()


@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])])
    async with await Config(descr) as cfg:
        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()
    assert not await list_sessions()


@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])
    async with await Config(od) as cfg:
        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']}
    assert not await list_sessions()


@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])
    async with await Config(od) as cfg:
        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).owner.get() == 'default'
        assert await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).owner.get() == 'default'
        assert await cfg.option('follower', 0).owner.get() == 'default'
        assert await cfg.option('follower', 1).owner.get() == 'default'
        assert await cfg.option('follower').owner.get() == ['default', 'default']
        #
        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
        assert await cfg.option('follower').value.get() == [None, 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'
        assert await cfg.option('follower').value.get() == [None, 'val3']
        #
        assert await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).owner.get() == 'default'
        assert await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).owner.get() == 'user'
        assert await cfg.option('follower', 0).owner.get() == 'default'
        assert await cfg.option('follower', 1).owner.get() == 'user'
        assert await cfg.option('follower').owner.get() == ['default', 'user']
    assert not await list_sessions()


#____________________________________________________________
@pytest.mark.asyncio
async def test_symlink_dependency():
    boolopt = BoolOption("b", "", default=False)
    linkopt = SymLinkOption("c", boolopt)
    descr = OptionDescription("opt", "",
                              [linkopt, OptionDescription("s1", "", [boolopt])])
    async with await Config(descr) as cfg:
        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
    assert not await list_sessions()


@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])])
    async with await Config(descr) as cfg:
        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}
    assert not await list_sessions()


@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])])
    async with await Config(descr) as cfg:
        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']
    assert not await list_sessions()