348 lines
15 KiB
Python
348 lines
15 KiB
Python
# 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).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'
|
|
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()
|