tiramisu/tests/test_symlink.py

348 lines
15 KiB
Python
Raw Normal View History

2013-08-24 22:32:54 +02:00
# coding: utf-8
2019-12-24 15:24:20 +01:00
import pytest
2017-07-09 09:49:03 +02:00
from .autopath import do_autopath
2015-07-24 17:54:10 +02:00
do_autopath()
2019-07-04 20:43:47 +02:00
from .config import config_type, get_config
2013-06-13 22:07:58 +02:00
2018-03-19 08:33:53 +01:00
from tiramisu import BoolOption, StrOption, SymLinkOption, \
2019-09-01 09:41:53 +02:00
OptionDescription, Leadership, Config, Calculation, calc_value, Params, ParamOption, ParamValue
2018-03-19 08:33:53 +01:00
from tiramisu.error import PropertiesOptionError, ConfigError
2013-06-13 22:07:58 +02:00
from tiramisu.setting import groups, owners
2018-10-31 08:00:19 +01:00
from tiramisu.storage import list_sessions
2020-01-22 20:46:18 +01:00
from .config import event_loop
2013-06-13 22:07:58 +02:00
def return_value():
pass
2013-06-13 22:07:58 +02:00
#____________________________________________________________
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_option(config_type):
2013-06-13 22:07:58 +02:00
boolopt = BoolOption("b", "", default=False)
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription("opt", "",
[linkopt, OptionDescription("s1", "", [boolopt])])
2020-01-22 20:46:18 +01:00
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()
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_assign_option(config_type):
2018-04-10 12:33:51 +02:00
boolopt = BoolOption("b", "", default=False)
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription("opt", "",
[linkopt, OptionDescription("s1", "", [boolopt])])
2020-01-22 20:46:18 +01:00
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()
2018-04-10 12:33:51 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_del_option(config_type):
2018-04-10 12:33:51 +02:00
boolopt = BoolOption("b", "", default=False)
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription("opt", "",
[linkopt, OptionDescription("s1", "", [boolopt])])
2020-01-22 20:46:18 +01:00
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()
2018-04-10 12:33:51 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_addproperties():
2018-04-11 16:36:15 +02:00
boolopt = BoolOption('b', '', default=True, properties=('test',))
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription('opt', '', [boolopt, linkopt])
2020-01-22 20:46:18 +01:00
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()
2018-04-11 16:36:15 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_getpermissive():
2018-09-15 22:44:49 +02:00
boolopt = BoolOption('b', '', default=True, properties=('test',))
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription('opt', '', [boolopt, linkopt])
2020-01-22 20:46:18 +01:00
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()
2018-09-15 22:44:49 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_addpermissives():
2018-04-11 16:36:15 +02:00
boolopt = BoolOption('b', '', default=True, properties=('test',))
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription('opt', '', [boolopt, linkopt])
2020-01-22 20:46:18 +01:00
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()
2018-04-11 16:36:15 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_getproperties():
boolopt = BoolOption('b', '', default=True, properties=('test',))
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription('opt', '', [boolopt, linkopt])
2020-01-22 20:46:18 +01:00
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()
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_getcallback():
2019-09-28 16:32:48 +02:00
boolopt = BoolOption('b', '', Calculation(return_value))
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription('opt', '', [boolopt, linkopt])
2020-01-22 20:46:18 +01:00
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()
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_requires(config_type):
2013-06-13 22:07:58 +02:00
boolopt = BoolOption('b', '', default=True)
2019-09-01 09:41:53 +02:00
disabled_property = Calculation(calc_value,
Params(ParamValue('disabled'),
kwargs={'condition': ParamOption(boolopt),
'expected': ParamValue(False)}))
stropt = StrOption('s', '', properties=(disabled_property,))
2013-06-13 22:07:58 +02:00
linkopt = SymLinkOption("c", stropt)
descr = OptionDescription('opt', '', [boolopt, stropt, linkopt])
2020-01-22 20:46:18 +01:00
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()
2013-06-13 22:07:58 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_multi(config_type):
2013-06-13 22:07:58 +02:00
boolopt = BoolOption("b", "", default=[False], multi=True)
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription("opt", "",
[linkopt, OptionDescription("s1", "", [boolopt])])
2020-01-22 20:46:18 +01:00
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()
2013-06-13 22:07:58 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_assign(config_type):
2019-07-04 20:43:47 +02:00
boolopt = BoolOption("b", "", default=False)
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription("opt", "",
[linkopt, OptionDescription("s1", "", [boolopt])])
2020-01-22 20:46:18 +01:00
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()
2018-03-19 08:33:53 +01:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_owner(config_type):
2013-06-13 22:07:58 +02:00
boolopt = BoolOption("b", "", default=False)
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription("opt", "",
[linkopt, OptionDescription("s1", "", [boolopt])])
2020-01-22 20:46:18 +01:00
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()
2013-06-13 22:07:58 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_get_information():
2013-06-13 22:07:58 +02:00
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'
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_leader():
2013-06-13 22:07:58 +02:00
a = StrOption('a', "", multi=True)
ip_admin_eth0 = SymLinkOption('ip_admin_eth0', a)
netmask_admin_eth0 = StrOption('netmask_admin_eth0', "", multi=True)
2020-01-22 20:46:18 +01:00
with pytest.raises(ValueError):
2019-12-24 15:24:20 +01:00
Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
2013-06-13 22:07:58 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_followers():
2013-06-13 22:07:58 +02:00
a = StrOption('a', "", multi=True)
ip_admin_eth0 = StrOption('ip_admin_eth0', "ip réseau autorisé", multi=True)
2013-06-13 22:07:58 +02:00
netmask_admin_eth0 = SymLinkOption('netmask_admin_eth0', a)
2020-01-22 20:46:18 +01:00
with pytest.raises(ValueError):
2019-12-24 15:24:20 +01:00
Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
2017-09-17 15:55:32 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_with_leader(config_type):
2018-04-19 08:19:03 +02:00
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)
2019-02-23 19:06:23 +01:00
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
leader = SymLinkOption('leader', ip_admin_eth0)
od = OptionDescription('root', '', [interface1, leader])
2020-01-22 20:46:18 +01:00
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()
2018-04-19 08:19:03 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_with_follower(config_type):
2018-04-19 08:19:03 +02:00
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)
2019-02-23 19:06:23 +01:00
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
follower = SymLinkOption('follower', netmask_admin_eth0)
od = OptionDescription('root', '', [interface1, follower])
2020-01-22 20:46:18 +01:00
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()
2018-04-19 08:19:03 +02:00
2017-09-17 15:55:32 +02:00
#____________________________________________________________
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_dependency():
2017-09-17 15:55:32 +02:00
boolopt = BoolOption("b", "", default=False)
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription("opt", "",
[linkopt, OptionDescription("s1", "", [boolopt])])
2020-01-22 20:46:18 +01:00
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()
2018-04-19 08:19:03 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_symlink_makedict(config_type):
2018-04-19 08:19:03 +02:00
boolopt = BoolOption("b", "", default=False)
linkopt = SymLinkOption("c", boolopt)
descr = OptionDescription("opt", "",
[linkopt, OptionDescription("s1", "", [boolopt])])
2020-01-22 20:46:18 +01:00
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()
2019-12-24 15:24:20 +01:00
@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])])
2020-01-22 20:46:18 +01:00
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()