1837 lines
100 KiB
Python
1837 lines
100 KiB
Python
# coding: utf-8
|
|
from .autopath import do_autopath
|
|
do_autopath()
|
|
import pytest
|
|
|
|
from tiramisu.setting import groups, owners
|
|
from tiramisu import BoolOption, StrOption, ChoiceOption, IPOption, \
|
|
NetworkOption, NetmaskOption, IntOption, FloatOption, \
|
|
StrOption, PortOption, BroadcastOption, DomainnameOption, \
|
|
EmailOption, URLOption, UsernameOption, FilenameOption, SymLinkOption, \
|
|
OptionDescription, DynOptionDescription, SynDynOption, submulti, Leadership, \
|
|
Config, \
|
|
Params, ParamOption, ParamValue, ParamSuffix, ParamSelfOption, ParamDynOption, ParamIndex, \
|
|
Calculation, calc_value, \
|
|
delete_session
|
|
from tiramisu.error import PropertiesOptionError, ConfigError, ConflictError
|
|
from tiramisu.storage import list_sessions
|
|
from .config import event_loop
|
|
|
|
|
|
class ConvertDynOptionDescription(DynOptionDescription):
|
|
def convert_suffix_to_path(self, suffix):
|
|
# remove dot with is illegal
|
|
return suffix.replace('.', '')
|
|
|
|
|
|
def return_true(value, param=None, suffix=None):
|
|
if value == 'val' and param in [None, 'yes']:
|
|
return
|
|
raise ValueError('no value')
|
|
|
|
|
|
def return_dynval(value='val', suffix=None):
|
|
return value
|
|
|
|
|
|
def return_list2(suffix):
|
|
return [str(suffix), 'val2']
|
|
|
|
|
|
def return_list(val=None, suffix=None):
|
|
if val:
|
|
return val
|
|
else:
|
|
return ['val1', 'val2']
|
|
|
|
|
|
def return_list_dot(val=None, suffix=None):
|
|
return ['val.1', 'val.2']
|
|
|
|
|
|
def return_same_list(*args, **kwargs):
|
|
return ['val1', 'val1']
|
|
|
|
|
|
def return_wrong_list(*args, **kwargs):
|
|
return ['---', ' ']
|
|
|
|
|
|
def return_raise(suffix):
|
|
raise Exception('error')
|
|
|
|
|
|
def return_str(*args, **kwargs):
|
|
return 'str'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_dyndescription():
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_list))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
async with await Config(od1) as cfg:
|
|
assert await cfg.value.dict() == {'dodval1.stval1': None, 'dodval2.stval2': None}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_dyndescription_raise():
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_raise))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
async with await Config(od1) as cfg:
|
|
with pytest.raises(ConfigError):
|
|
await cfg.value.dict()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_dyndescription_not_list():
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_str))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
async with await Config(od1) as cfg:
|
|
with pytest.raises(ValueError):
|
|
await cfg.value.dict()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subpath_dyndescription():
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_list))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od1])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.value.dict() == {'od.dodval1.stval1': None, 'od.dodval2.stval2': None}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_dyndescription():
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_dyndescription():
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval3').value.get()
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval1.novalue').value.get()
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval1.stnoval1').value.get()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_getdoc_dyndescription():
|
|
st1 = StrOption('st', 'doc1')
|
|
dod = DynOptionDescription('dod', 'doc2', [st1], suffixes=Calculation(return_list))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od1])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.option('od.dodval1.stval1').option.name() == 'stval1'
|
|
assert await cfg.option('od.dodval2.stval2').option.name() == 'stval2'
|
|
assert await cfg.option('od.dodval1').option.name() == 'dodval1'
|
|
assert await cfg.option('od.dodval2').option.name() == 'dodval2'
|
|
assert await cfg.option('od.dodval1.stval1').option.doc() == 'doc1val1'
|
|
assert await cfg.option('od.dodval2.stval2').option.doc() == 'doc1val2'
|
|
assert await cfg.option('od.dodval1').option.doc() == 'doc2val1'
|
|
assert await cfg.option('od.dodval2').option.doc() == 'doc2val2'
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mod_dyndescription():
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
#
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.dodval1.stval1').value.set('yes')
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'yes'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.dodval2.stval2').value.set('no')
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'yes'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'no'
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.get() == owner
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_del_dyndescription():
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
await cfg.option('od.dodval1.stval1').value.set('yes')
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
await cfg.option('od.dodval1.stval1').value.reset()
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multi_dyndescription():
|
|
st = StrOption('st', '', multi=True)
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == []
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == []
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval1.stval1').value.set(['yes'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['yes']
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == []
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval2.stval2').value.set(['no'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['yes']
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == ['no']
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.get() == owner
|
|
await cfg.option('od.dodval1.stval1').value.set(['yes', 'no'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['yes', 'no']
|
|
await cfg.option('od.dodval1.stval1').value.set(['yes'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['yes']
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prop_dyndescription():
|
|
st = StrOption('st', '', properties=('test',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
assert set(await cfg.option('od.dodval1.stval1').property.get()) == set(['test'])
|
|
assert set(await cfg.option('od.dodval2.stval2').property.get()) == set(['test'])
|
|
await cfg.option('od.dodval2.stval2').property.add('test2')
|
|
assert set(await cfg.option('od.dodval1.stval1').property.get()) == set(['test'])
|
|
assert set(await cfg.option('od.dodval2.stval2').property.get()) == set(['test', 'test2'])
|
|
await cfg.option('od.dodval1.stval1').property.pop('test')
|
|
assert set(await cfg.option('od.dodval1.stval1').property.get()) == set([])
|
|
#
|
|
assert set(await cfg.option('od.dodval1').property.get()) == set([])
|
|
assert set(await cfg.option('od.dodval2').property.get()) == set([])
|
|
await cfg.option('od.dodval1').property.add('test1')
|
|
assert set(await cfg.option('od.dodval1').property.get()) == set(['test1'])
|
|
assert set(await cfg.option('od.dodval2').property.get()) == set([])
|
|
await cfg.option('od.dodval1').property.pop('test1')
|
|
assert set(await cfg.option('od.dodval1').property.get()) == set([])
|
|
assert set(await cfg.option('od.dodval2').property.get()) == set([])
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prop_dyndescription_force_store_value():
|
|
st = StrOption('st', '', properties=('force_store_value',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_write()
|
|
assert await cfg.value.dict() == {'od.dodval1.stval1': None, 'od.dodval2.stval2': None}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prop_dyndescription_force_store_value_calculation_prefix():
|
|
lst = StrOption('lst', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', Calculation(return_list, Params(ParamSuffix())) , properties=('force_store_value',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(lst))))
|
|
od = OptionDescription('od', '', [dod, lst])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_write()
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault() == False
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault() == False
|
|
assert await cfg.value.dict() == {'od.lst': ['val1', 'val2'], 'od.dodval1.stval1': 'val1', 'od.dodval2.stval2': 'val2'}
|
|
#
|
|
await cfg.option('od.lst').value.set(['val1', 'val2', 'val3'])
|
|
assert await cfg.option('od.dodval3.stval3').owner.isdefault() == False
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault() == False
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault() == False
|
|
assert await cfg.value.dict() == {'od.lst': ['val1', 'val2', 'val3'], 'od.dodval1.stval1': 'val1', 'od.dodval2.stval2': 'val2', 'od.dodval3.stval3': 'val3'}
|
|
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_dyndescription():
|
|
st = StrOption('st', '', Calculation(return_dynval))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'val'
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval1.stval1').value.set('val2')
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val2'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'val'
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval1.stval1').value.reset()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'val'
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_dyndescription_outside_wrong_param():
|
|
lst = StrOption('lst', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', Calculation(return_dynval))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(lst))))
|
|
out = StrOption('out', '', Calculation(return_dynval, Params(ParamOption(st))))
|
|
od = OptionDescription('od', '', [dod, out])
|
|
od2 = OptionDescription('od', '', [od, lst])
|
|
async with await Config(od2) as cfg:
|
|
with pytest.raises(ConfigError):
|
|
await cfg.value.dict()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_dyndescription_outside1():
|
|
lst = StrOption('lst', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', Calculation(return_dynval))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(lst))))
|
|
out = StrOption('out', '', Calculation(return_dynval, Params(ParamDynOption(st, 'val1', dod))))
|
|
od = OptionDescription('od', '', [dod, out])
|
|
od2 = OptionDescription('od', '', [od, lst])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.value.dict() == {'od.dodval1.stval1': 'val', 'od.dodval2.stval2': 'val', 'od.out': 'val', 'lst': ['val1', 'val2']}
|
|
await cfg.option('od.dodval1.stval1').value.set('val1')
|
|
await cfg.option('od.dodval2.stval2').value.set('val2')
|
|
assert await cfg.value.dict() == {'od.dodval1.stval1': 'val1', 'od.dodval2.stval2': 'val2', 'od.out': 'val1', 'lst': ['val1', 'val2']}
|
|
await cfg.option('lst').value.set(['val2'])
|
|
with pytest.raises(ConfigError):
|
|
await cfg.value.dict()
|
|
await cfg.option('lst').value.set(['val1'])
|
|
assert await cfg.value.dict() == {'od.dodval1.stval1': 'val1', 'od.out': 'val1', 'lst': ['val1']}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_dyndescription_outside2():
|
|
lst = StrOption('lst', '', ['val1', 'val2'], multi=True)
|
|
out = StrOption('out', '')
|
|
st = StrOption('st', '', Calculation(return_dynval, Params(ParamOption(out))))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(lst))))
|
|
od = OptionDescription('od', '', [dod, out])
|
|
od2 = OptionDescription('od', '', [od, lst])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.value.dict() == {'od.dodval1.stval1': None, 'od.dodval2.stval2': None, 'od.out': None, 'lst': ['val1', 'val2']}
|
|
await cfg.option('od.out').value.set('val1')
|
|
assert await cfg.value.dict() == {'od.dodval1.stval1': 'val1', 'od.dodval2.stval2': 'val1', 'od.out': 'val1', 'lst': ['val1', 'val2']}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_list_dyndescription():
|
|
st = StrOption('st', '', Calculation(return_list2, Params(ParamSuffix())), multi=True, properties=('notunique',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['val1', 'val2']
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == ['val2', 'val2']
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval1.stval1').value.set(['val3', 'val2'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['val3', 'val2']
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == ['val2', 'val2']
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mandatory_dyndescription():
|
|
st = StrOption('st', '', properties=('mandatory',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_only()
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
await cfg.property.read_write()
|
|
await cfg.option('od.dodval1.stval1').value.set('val')
|
|
await cfg.property.read_only()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val'
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
await cfg.property.read_write()
|
|
await cfg.option('od.dodval1.stval1').value.reset()
|
|
await cfg.property.read_only()
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
assert list(await cfg.value.mandatory()) == ['od.dodval1.stval1', 'od.dodval2.stval2']
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od1 = OptionDescription('od', '', [dod, val1])
|
|
async with await Config(od1) as cfg:
|
|
assert await cfg.value.dict() == {'dodval1.stval1': None, 'dodval2.stval2': None, 'val1': ['val1', 'val2']}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subpath_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od1 = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od1])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.value.dict() == {'od.dodval1.stval1': None, 'od.dodval2.stval2': None, 'od.val1': ['val1', 'val2']}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval3').value.get()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mod_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval1.stval1').value.set('yes')
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'yes'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval2.stval2').value.set('no')
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'yes'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'no'
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.get() == owner
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_del_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
await cfg.option('od.dodval1.stval1').value.set('yes')
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
await cfg.option('od.dodval1.stval1').value.reset()
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multi_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', multi=True, properties=('notunique',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == []
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == []
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval1.stval1').value.set(['yes'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['yes']
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == []
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval2.stval2').value.set(['no'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['yes']
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == ['no']
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.get() == owner
|
|
await cfg.option('od.dodval1.stval1').value.set(['yes', 'yes'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['yes', 'yes']
|
|
await cfg.option('od.dodval1.stval1').value.set(['yes'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == ['yes']
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prop_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', properties=('test',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
assert set(await cfg.option('od.dodval1.stval1').property.get()) == set(['test'])
|
|
assert set(await cfg.option('od.dodval2.stval2').property.get()) == set(['test'])
|
|
await cfg.option('od.dodval2.stval2').property.add('test2')
|
|
assert set(await cfg.option('od.dodval1.stval1').property.get()) == set(['test'])
|
|
assert set(await cfg.option('od.dodval2.stval2').property.get()) == set(['test', 'test2'])
|
|
await cfg.option('od.dodval1.stval1').property.pop('test')
|
|
assert set(await cfg.option('od.dodval1.stval1').property.get()) == set([])
|
|
assert set(await cfg.option('od.dodval2.stval2').property.get()) == set(['test', 'test2'])
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', Calculation(return_dynval))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'val'
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval1.stval1').value.set('val2')
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val2'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'val'
|
|
assert await cfg.option('od.dodval1.stval1').owner.get() == owner
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
await cfg.option('od.dodval1.stval1').value.reset()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val'
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'val'
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mandatory_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', properties=('mandatory',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_only()
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
await cfg.property.read_write()
|
|
await cfg.option('od.dodval1.stval1').value.set('val')
|
|
await cfg.property.read_only()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val'
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
await cfg.property.read_write()
|
|
await cfg.option('od.dodval1.stval1').value.reset()
|
|
await cfg.property.read_only()
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
assert list(await cfg.value.mandatory()) == ['od.dodval1.stval1', 'od.dodval2.stval2']
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_increase_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', properties=('mandatory',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_write()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval3.stval3').value.get()
|
|
await cfg.option('od.val1').value.set(['val1', 'val2', 'val3'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
assert await cfg.option('od.dodval3.stval3').value.get() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decrease_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', properties=('mandatory',))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
await cfg.property.read_write()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
await cfg.option('od.dodval2.stval2').value.set('yes')
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() == 'yes'
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
assert await cfg.option('od.dodval2.stval2').owner.get() == owner
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval3').value.get()
|
|
await cfg.option('od.val1').value.set(['val1'])
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval2').value.get()
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval3').value.get()
|
|
assert await cfg.option('od.dodval1.stval1').owner.isdefault()
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval2.stval2').owner.get()
|
|
with pytest.raises(AttributeError):
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dyndescription_root():
|
|
boolean = BoolOption('boolean', '', True)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
st1 = StrOption('st', '', properties=(disabled_property,))
|
|
dod = DynOptionDescription('dod', '', [boolean, st1], suffixes=Calculation(return_list))
|
|
with pytest.raises(ConfigError):
|
|
await Config(dod, session_id='error')
|
|
await delete_session('error')
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dyndescription_disable_suffix_root():
|
|
boolean = BoolOption('boolean', '', True)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
val = StrOption('val', '', ['ext1', 'ext2'], properties=(disabled_property,), multi=True)
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(calc_value, Params(ParamOption(val, notraisepropertyerror=True))), properties=(disabled_property,))
|
|
od = OptionDescription('root', 'root', [boolean, val, dod])
|
|
config = await Config(od, session_id='test')
|
|
await config.property.read_write()
|
|
assert await config.value.dict() == {'boolean': True, 'val': ['ext1', 'ext2'], 'dodext1.stext1': None, 'dodext2.stext2': None}
|
|
#
|
|
await config.option('boolean').value.set(False)
|
|
assert await config.value.dict() == {'boolean': False}
|
|
await delete_session('test')
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dyndescription_disable_suffix_root_2():
|
|
boolean = BoolOption('boolean', '', False)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
val = StrOption('val', '', ['ext1', 'ext2'], properties=(disabled_property,), multi=True)
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(calc_value, Params(ParamOption(val, notraisepropertyerror=True))), properties=(disabled_property,))
|
|
od = OptionDescription('root', 'root', [boolean, val, dod])
|
|
config = await Config(od, session_id='test')
|
|
await config.property.read_write()
|
|
assert await config.value.dict() == {'boolean': False}
|
|
#
|
|
await config.option('boolean').value.set(True)
|
|
assert await config.value.dict() == {'boolean': True, 'val': ['ext1', 'ext2'], 'dodext1.stext1': None, 'dodext2.stext2': None}
|
|
await delete_session('test')
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dyndescription_disable_suffix():
|
|
boolean = BoolOption('boolean', '', True)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
val = StrOption('val', '', ['ext1', 'ext2'], properties=(disabled_property,), multi=True)
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(calc_value, Params(ParamOption(val, notraisepropertyerror=True))), properties=(disabled_property,))
|
|
od = OptionDescription('root', 'root', [boolean, val, dod])
|
|
od2 = OptionDescription('root', 'root', [od])
|
|
config = await Config(od2, session_id='test')
|
|
await config.property.read_write()
|
|
assert await config.value.dict() == {'root.boolean': True, 'root.val': ['ext1', 'ext2'], 'root.dodext1.stext1': None, 'root.dodext2.stext2': None}
|
|
#
|
|
await config.option('root.boolean').value.set(False)
|
|
assert await config.value.dict() == {'root.boolean': False}
|
|
await delete_session('test')
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_requires_dyndescription():
|
|
boolean = BoolOption('boolean', '', True)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean, raisepropertyerror=True),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
st1 = StrOption('st', '', properties=(disabled_property,))
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_list))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od1, boolean])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_write()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
#
|
|
await cfg.option('boolean').value.set(False)
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert frozenset(props) == frozenset(['disabled'])
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert frozenset(props) == frozenset(['disabled'])
|
|
#
|
|
await cfg.option('boolean').value.set(True)
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
#transitive
|
|
await cfg.option('boolean').property.add('disabled')
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert frozenset(props) == frozenset(['disabled'])
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert frozenset(props) == frozenset(['disabled'])
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_requires_dyndescription_boolean():
|
|
boolean1 = BoolOption('boolean1', '', True)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean1, raisepropertyerror=True),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
boolean = BoolOption('boolean', '', True, properties=(disabled_property,))
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean, raisepropertyerror=True),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
st = StrOption('st', '', properties=(disabled_property,))
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od, boolean1, boolean])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_write()
|
|
assert await cfg.value.dict() == {'boolean1': True,
|
|
'boolean': True,
|
|
'od.dodval1.stval1': None,
|
|
'od.dodval2.stval2': None}
|
|
#
|
|
await cfg.option('boolean').value.set(False)
|
|
assert await cfg.value.dict() == {'boolean1': True,
|
|
'boolean': False}
|
|
#
|
|
await cfg.option('boolean').value.set(True)
|
|
assert await cfg.value.dict() == {'boolean1': True,
|
|
'boolean': True,
|
|
'od.dodval1.stval1': None,
|
|
'od.dodval2.stval2': None}
|
|
#
|
|
await cfg.option('boolean1').value.set(False)
|
|
assert await cfg.value.dict() == {'boolean1': False}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_requires_dyndescription_in_dyn():
|
|
boolean = BoolOption('boolean', '', True)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean, raisepropertyerror=True),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
st = StrOption('st', '', properties=(disabled_property,))
|
|
dod = DynOptionDescription('dod', '', [boolean, st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_write()
|
|
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
#
|
|
await cfg.option('od.dodval1.booleanval1').value.set(False)
|
|
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert props == frozenset(['disabled'])
|
|
props = []
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
#
|
|
await cfg.option('od.dodval1.booleanval1').value.set(True)
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_requires_dyndescription2():
|
|
boolean = BoolOption('boolean', '', True)
|
|
st1 = StrOption('st', '')
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(boolean, raisepropertyerror=True),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_list), properties=(disabled_property,))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
od2 = OptionDescription('od', '', [od1, boolean])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.property.read_write()
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
#
|
|
await cfg.option('boolean').value.set(False)
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert frozenset(props) == frozenset(['disabled'])
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert frozenset(props) == frozenset(['disabled'])
|
|
#
|
|
await cfg.option('boolean').value.set(True)
|
|
assert await cfg.option('od.dodval1.stval1').value.get() is None
|
|
assert await cfg.option('od.dodval2.stval2').value.get() is None
|
|
#transitive
|
|
await cfg.option('boolean').property.add('disabled')
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval1.stval1').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert frozenset(props) == frozenset(['disabled'])
|
|
props = []
|
|
try:
|
|
await cfg.option('od.dodval2.stval2').value.get()
|
|
except PropertiesOptionError as err:
|
|
props = err.proptype
|
|
assert frozenset(props) == frozenset(['disabled'])
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validator_dyndescription():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '', validators=[Calculation(return_true, Params((ParamSelfOption(), ParamValue('yes'))))], default='val')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.option('od.dodval1.stval1').value.get() == 'val'
|
|
with pytest.raises(ValueError):
|
|
await cfg.option('od.dodval1.stval1').value.set('no')
|
|
await cfg.option('od.dodval1.stval1').value.set('val')
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_makedict_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.option('od.dodval1.stval1').value.set('yes')
|
|
assert await cfg.value.dict() == {'od.val1': ['val1', 'val2'], 'od.dodval1.stval1': 'yes', 'od.dodval2.stval2': None}
|
|
assert await cfg.value.dict(flatten=True) == {'val1': ['val1', 'val2'], 'stval1': 'yes', 'stval2': None}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await cfg.option('od.dodval1.stval1').value.set('yes')
|
|
ret = await cfg.option.find('stval1', first=True)
|
|
assert await ret.value.get() == "yes"
|
|
ret = await cfg.option.find('stval1', first=True)
|
|
assert isinstance(await ret.option.get(), SynDynOption)
|
|
#assert await cfg.option.find(bytype=StrOption, type='path') == ['od.dodval1.stval1', 'od.dodval2.stval2', 'od.val1']
|
|
#opts = await cfg.option.find(byvalue='yes')
|
|
#assert len(opts) == 1
|
|
#assert isinstance(opts[0], SynDynOption)
|
|
#assert opts[0].impl_getname() == 'stval1'
|
|
with pytest.raises(AttributeError):
|
|
list(await cfg.option.find('strnotexists'))
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_information_dyndescription_context():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod, val1])
|
|
od2 = OptionDescription('od', '', [od])
|
|
dod.impl_set_information('testod', 'val1')
|
|
st.impl_set_information('testst', 'val2')
|
|
async with await Config(od2) as cfg:
|
|
await cfg.information.set('testcfgod', 'val3')
|
|
assert await cfg.option('od.dodval1').information.get('testod') == 'val1'
|
|
assert await cfg.option('od.dodval2').information.get('testod') == 'val1'
|
|
assert await cfg.option('od.dodval1.stval1').information.get('testst') == 'val2'
|
|
assert await cfg.option('od.dodval2.stval2').information.get('testst') == 'val2'
|
|
assert await cfg.information.get('testcfgod') == 'val3'
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_dyndescription():
|
|
st = StrOption('st', '')
|
|
ip = IPOption('ip', '')
|
|
network = NetworkOption('network', '')
|
|
netmask = NetmaskOption('netmask', '')
|
|
ch = ChoiceOption('ch', '', ('val1', 'val2', 'val3'))
|
|
ch1 = ChoiceOption('ch1', '', Calculation(return_list))
|
|
boo = BoolOption('boo', '')
|
|
intr = IntOption('intr', '')
|
|
floa = FloatOption('floa', '')
|
|
uni = StrOption('uni', '')
|
|
port = PortOption('port', '')
|
|
broad = BroadcastOption('broad', '')
|
|
domain = DomainnameOption('domain', '')
|
|
email = EmailOption('email', '')
|
|
url = URLOption('url', '')
|
|
username = UsernameOption('username', '')
|
|
filename = FilenameOption('filename', '')
|
|
dod = DynOptionDescription('dod', '', [st, ip, network, netmask, ch, ch1,
|
|
boo, intr, floa, uni, port, broad,
|
|
domain, email, url, username,
|
|
filename], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [dod])
|
|
async with await Config(od) as cfg:
|
|
assert await cfg.option('dodval1.stval1').value.get() is None
|
|
assert await cfg.option('dodval1.ipval1').value.get() is None
|
|
assert await cfg.option('dodval1.networkval1').value.get() is None
|
|
assert await cfg.option('dodval1.netmaskval1').value.get() is None
|
|
assert await cfg.option('dodval1.chval1').value.get() is None
|
|
assert await cfg.option('dodval1.ch1val1').value.get() is None
|
|
assert await cfg.option('dodval1.booval1').value.get() is None
|
|
assert await cfg.option('dodval1.intrval1').value.get() is None
|
|
assert await cfg.option('dodval1.floaval1').value.get() is None
|
|
assert await cfg.option('dodval1.unival1').value.get() is None
|
|
assert await cfg.option('dodval1.portval1').value.get() is None
|
|
assert await cfg.option('dodval1.broadval1').value.get() is None
|
|
assert await cfg.option('dodval1.domainval1').value.get() is None
|
|
assert await cfg.option('dodval1.emailval1').value.get() is None
|
|
assert await cfg.option('dodval1.urlval1').value.get() is None
|
|
assert await cfg.option('dodval1.usernameval1').value.get() is None
|
|
assert await cfg.option('dodval1.filenameval1').value.get() is None
|
|
#
|
|
await cfg.option('dodval1.stval1').value.set("no")
|
|
await cfg.option('dodval1.ipval1').value.set("1.1.1.1")
|
|
await cfg.option('dodval1.networkval1').value.set("1.1.1.0")
|
|
await cfg.option('dodval1.netmaskval1').value.set("255.255.255.0")
|
|
await cfg.option('dodval1.chval1').value.set("val1")
|
|
await cfg.option('dodval1.ch1val1').value.set("val2")
|
|
await cfg.option('dodval1.booval1').value.set(True)
|
|
await cfg.option('dodval1.intrval1').value.set(1)
|
|
await cfg.option('dodval1.floaval1').value.set(0.1)
|
|
await cfg.option('dodval1.unival1').value.set(u"no")
|
|
await cfg.option('dodval1.portval1').value.set('80')
|
|
await cfg.option('dodval1.broadval1').value.set("1.1.1.255")
|
|
await cfg.option('dodval1.domainval1').value.set("test.com")
|
|
await cfg.option('dodval1.emailval1').value.set("test@test.com")
|
|
await cfg.option('dodval1.urlval1').value.set("http://test.com")
|
|
await cfg.option('dodval1.usernameval1').value.set("user1")
|
|
await cfg.option('dodval1.filenameval1').value.set("/tmp")
|
|
assert await cfg.option('dodval1.stval1').value.get() == "no"
|
|
assert await cfg.option('dodval1.ipval1').value.get() == "1.1.1.1"
|
|
assert await cfg.option('dodval1.networkval1').value.get() == "1.1.1.0"
|
|
assert await cfg.option('dodval1.netmaskval1').value.get() == "255.255.255.0"
|
|
assert await cfg.option('dodval1.chval1').value.get() == "val1"
|
|
assert await cfg.option('dodval1.ch1val1').value.get() == "val2"
|
|
assert await cfg.option('dodval1.booval1').value.get() is True
|
|
assert await cfg.option('dodval1.intrval1').value.get() == 1
|
|
assert await cfg.option('dodval1.floaval1').value.get() == 0.1
|
|
assert await cfg.option('dodval1.unival1').value.get() == u"no"
|
|
assert await cfg.option('dodval1.portval1').value.get() == '80'
|
|
assert await cfg.option('dodval1.broadval1').value.get() == "1.1.1.255"
|
|
assert await cfg.option('dodval1.domainval1').value.get() == "test.com"
|
|
assert await cfg.option('dodval1.emailval1').value.get() == "test@test.com"
|
|
assert await cfg.option('dodval1.urlval1').value.get() == "http://test.com"
|
|
assert await cfg.option('dodval1.usernameval1').value.get() == "user1"
|
|
assert await cfg.option('dodval1.filenameval1').value.get() == "/tmp"
|
|
assert await cfg.option('dodval2.stval2').value.get() is None
|
|
assert await cfg.option('dodval2.ipval2').value.get() is None
|
|
assert await cfg.option('dodval2.networkval2').value.get() is None
|
|
assert await cfg.option('dodval2.netmaskval2').value.get() is None
|
|
assert await cfg.option('dodval2.chval2').value.get() is None
|
|
assert await cfg.option('dodval2.ch1val2').value.get() is None
|
|
assert await cfg.option('dodval2.booval2').value.get() is None
|
|
assert await cfg.option('dodval2.intrval2').value.get() is None
|
|
assert await cfg.option('dodval2.floaval2').value.get() is None
|
|
assert await cfg.option('dodval2.unival2').value.get() is None
|
|
assert await cfg.option('dodval2.portval2').value.get() is None
|
|
assert await cfg.option('dodval2.broadval2').value.get() is None
|
|
assert await cfg.option('dodval2.domainval2').value.get() is None
|
|
assert await cfg.option('dodval2.emailval2').value.get() is None
|
|
assert await cfg.option('dodval2.urlval2').value.get() is None
|
|
assert await cfg.option('dodval2.usernameval2').value.get() is None
|
|
assert await cfg.option('dodval2.filenameval2').value.get() is None
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [st])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
#
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [None], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': ['yes']}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == None
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('no')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.pop(0)
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_default_multi_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True, default_multi='no')
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [st])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
#
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_dyndescription_param():
|
|
val1 = StrOption('val1', '', ['val1', 'val2'], multi=True)
|
|
odval = OptionDescription('odval1', '', [val1])
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [st, odval])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': [], 'od.odval1.val1': ['val1', 'val2']}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owners.default
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [None], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': ['yes'], 'od.odval1.val1': ['val1', 'val2']}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == None
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owners.default
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('no')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.pop(0)
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owners.default
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owners.default
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_default_multi_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True, default_multi='no')
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [st])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
#
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
async def _test_leadership(cfg):
|
|
owner = await cfg.owner.get()
|
|
await cfg.option('od.val1.val1').value.set(['val1', 'val2'])
|
|
await cfg.option('od.val1.val2', 0).value.set('val1')
|
|
await cfg.option('od.val1.val2', 1).value.set('val2')
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': [], 'od.val1.val1': ['val1', 'val2'], 'od.val1.val2': ['val1', 'val2']}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owners.default
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [None], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': ['yes'], 'od.val1.val1': ['val1', 'val2'], 'od.val1.val2': ['val1', 'val2']}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == None
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owners.default
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('no')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.pop(0)
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owners.default
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owners.default
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.get() == owners.default
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_dyndescription_param_leader():
|
|
val1 = StrOption('val1', "", multi=True)
|
|
val2 = StrOption('val2', "", multi=True)
|
|
odval = Leadership('val1', '', [val1, val2])
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list, Params(ParamOption(val1))))
|
|
od = OptionDescription('od', '', [st, odval])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await _test_leadership(cfg)
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_default_multi_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True, default_multi='no')
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [st])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
#
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_dyndescription_param_follower():
|
|
val1 = StrOption('val1', "", multi=True)
|
|
val2 = StrOption('val2', "", multi=True)
|
|
odval = Leadership('val1', '', [val1, val2])
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list, Params(ParamOption(val2))))
|
|
od = OptionDescription('od', '', [st, odval])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
await _test_leadership(cfg)
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_default_multi_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True, default_multi='no')
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [st])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
#
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_submulti_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=submulti)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
std = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od1 = OptionDescription('od', '', [std])
|
|
od2 = OptionDescription('od', '', [od1])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
#
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set(['no'])
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == ['no']
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_callback_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", Calculation(return_dynval, Params(kwargs={'value': ParamOption(st1)})), multi=True)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st1 = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od1 = OptionDescription('od', '', [st1])
|
|
od2 = OptionDescription('od', '', [od1])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() ==[]
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': ['yes'], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': ['yes']}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'yes'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('no')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.pop(0)
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'yes'
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_callback_value_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", Calculation(return_dynval, Params(kwargs={'value': ParamValue('val')})), multi=True)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od = OptionDescription('od', '', [st])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('val')
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'val'
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_callback_nomulti_dyndescription():
|
|
v11 = StrOption('v1', '', "val")
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", Calculation(return_dynval, Params(ParamOption(v11))), multi=True)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
stt = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od1 = OptionDescription('od', '', [stt])
|
|
od2 = OptionDescription('od', '', [od1, v11])
|
|
async with await Config(od2) as cfg:
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'val'
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_callback_samegroup_dyndescription():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True)
|
|
st3 = StrOption('st3', "", Calculation(return_dynval, Params(ParamOption(st2))), multi=True)
|
|
stm = Leadership('st1', '', [st1, st2, st3])
|
|
stt = DynOptionDescription('st', '', [stm], suffixes=Calculation(return_list))
|
|
od1 = OptionDescription('od', '', [stt])
|
|
od2 = OptionDescription('od', '', [od1])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st1val1': [],
|
|
'od.stval1.st1val1.st2val1': [],
|
|
'od.stval1.st1val1.st3val1': [],
|
|
'od.stval2.st1val2.st1val2': [],
|
|
'od.stval2.st1val2.st2val2': [],
|
|
'od.stval2.st1val2.st3val2': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st1val1': ['yes'],
|
|
'od.stval1.st1val1.st2val1': [None],
|
|
'od.stval1.st1val1.st3val1': [None],
|
|
'od.stval2.st1val2.st1val2': [],
|
|
'od.stval2.st1val2.st2val2': [],
|
|
'od.stval2.st1val2.st3val2': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval1.st1val1.st3val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st1val1': ['yes'],
|
|
'od.stval1.st1val1.st2val1': ['yes'],
|
|
'od.stval1.st1val1.st3val1': ['yes'],
|
|
'od.stval2.st1val2.st1val2': [],
|
|
'od.stval2.st1val2.st2val2': [],
|
|
'od.stval2.st1val2.st3val2': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st3val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_conflict_dyndescription():
|
|
st = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st], suffixes=Calculation(return_list))
|
|
dodinvalid = StrOption('dodinvalid', '')
|
|
dod, dodinvalid
|
|
with pytest.raises(ConflictError):
|
|
OptionDescription('od', '', [dod, dodinvalid])
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_subod_dyndescription():
|
|
st2 = StrOption('st2', '')
|
|
od1 = OptionDescription('od1', '', [st2])
|
|
od1
|
|
with pytest.raises(ConfigError):
|
|
DynOptionDescription('dod', '', [od1], suffixes=Calculation(return_list))
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_subdynod_dyndescription():
|
|
st2 = StrOption('st2', '')
|
|
od1 = DynOptionDescription('od1', '', [st2], suffixes=Calculation(return_list))
|
|
od1
|
|
with pytest.raises(ConfigError):
|
|
DynOptionDescription('dod', '', [od1], suffixes=Calculation(return_list))
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_symlink_dyndescription():
|
|
st = StrOption('st', '')
|
|
st2 = SymLinkOption('st2', st)
|
|
st2
|
|
with pytest.raises(ConfigError):
|
|
DynOptionDescription('dod', '', [st, st2], suffixes=Calculation(return_list))
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nocallback_dyndescription():
|
|
st = StrOption('st', '')
|
|
st2 = StrOption('st2', '')
|
|
st, st2
|
|
with pytest.raises(TypeError):
|
|
DynOptionDescription('dod', '', [st, st2])
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_samevalue_dyndescription():
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_same_list))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
async with await Config(od1) as cfg:
|
|
with pytest.raises(ValueError):
|
|
await cfg.value.dict()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_name_dyndescription():
|
|
st1 = StrOption('st', '')
|
|
dod = DynOptionDescription('dod', '', [st1], suffixes=Calculation(return_wrong_list))
|
|
od1 = OptionDescription('od', '', [dod])
|
|
async with await Config(od1) as cfg:
|
|
with pytest.raises(ValueError):
|
|
await cfg.value.dict()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_dyndescription_convert():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True)
|
|
stm = Leadership('st1', '', [st1, st2])
|
|
st = ConvertDynOptionDescription('st', '', [stm], suffixes=Calculation(return_list_dot))
|
|
od = OptionDescription('od', '', [st])
|
|
od2 = OptionDescription('od', '', [od])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
#
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st2val1': [None], 'od.stval2.st1val2.st2val2': [], 'od.stval2.st1val2.st1val2': [], 'od.stval1.st1val1.st1val1': ['yes']}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == None
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('no')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == ['yes']
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).value.get() == 'no'
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.pop(0)
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.reset()
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leadership_callback_samegroup_dyndescription_convert():
|
|
st1 = StrOption('st1', "", multi=True)
|
|
st2 = StrOption('st2', "", multi=True)
|
|
st3 = StrOption('st3', "", Calculation(return_dynval, Params(ParamOption(st2))), multi=True)
|
|
stm = Leadership('st1', '', [st1, st2, st3])
|
|
stt = ConvertDynOptionDescription('st', '', [stm], suffixes=Calculation(return_list_dot))
|
|
od1 = OptionDescription('od', '', [stt])
|
|
od2 = OptionDescription('od', '', [od1])
|
|
async with await Config(od2) as cfg:
|
|
owner = await cfg.owner.get()
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st1val1': [],
|
|
'od.stval1.st1val1.st2val1': [],
|
|
'od.stval1.st1val1.st3val1': [],
|
|
'od.stval2.st1val2.st1val2': [],
|
|
'od.stval2.st1val2.st2val2': [],
|
|
'od.stval2.st1val2.st3val2': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').value.get() == []
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').value.get() == []
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st1val1').value.set(['yes'])
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st1val1': ['yes'],
|
|
'od.stval1.st1val1.st2val1': [None],
|
|
'od.stval1.st1val1.st3val1': [None],
|
|
'od.stval2.st1val2.st1val2': [],
|
|
'od.stval2.st1val2.st2val2': [],
|
|
'od.stval2.st1val2.st3val2': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval1.st1val1.st3val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
#
|
|
await cfg.option('od.stval1.st1val1.st2val1', 0).value.set('yes')
|
|
assert await cfg.value.dict() == {'od.stval1.st1val1.st1val1': ['yes'],
|
|
'od.stval1.st1val1.st2val1': ['yes'],
|
|
'od.stval1.st1val1.st3val1': ['yes'],
|
|
'od.stval2.st1val2.st1val2': [],
|
|
'od.stval2.st1val2.st2val2': [],
|
|
'od.stval2.st1val2.st3val2': []}
|
|
assert await cfg.option('od.stval1.st1val1.st1val1').owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st2val1', 0).owner.get() == owner
|
|
assert await cfg.option('od.stval1.st1val1.st3val1', 0).owner.isdefault()
|
|
assert await cfg.option('od.stval2.st1val2.st1val2').owner.isdefault()
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dyn_with_leader_hidden_in_config():
|
|
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, properties=('hidden',))
|
|
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0], properties=('hidden',))
|
|
dyn = DynOptionDescription('leader', '', [interface1], suffixes=Calculation(return_list))
|
|
od = OptionDescription('root', '', [dyn])
|
|
async with await Config(od) as cfg:
|
|
await cfg.property.read_write()
|
|
await cfg.permissive.add('hidden')
|
|
assert await cfg.forcepermissive.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.get() == []
|
|
await cfg.forcepermissive.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.set(['192.168.1.1'])
|
|
assert await cfg.forcepermissive.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 0).value.get() is None
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.get()
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 0).value.get()
|
|
await cfg.value.dict(leader_to_list=True) == {'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': [{'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': '192.168.1.1'}],
|
|
'leaderval2.ip_admin_eth0val2.ip_admin_eth0val2': [{'leaderval2.ip_admin_eth0val2.ip_admin_eth0val2': '192.168.1.1'}]}
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dyn_leadership_requires():
|
|
ip_admin_eth0 = StrOption('ip_admin_eth0', "ip réseau autorisé", multi=True, properties=('notunique',))
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(ip_admin_eth0, notraisepropertyerror=True),
|
|
'expected': ParamValue('192.168.1.1'),
|
|
'no_condition_is_invalid': ParamValue(True),
|
|
'index': ParamIndex()}))
|
|
netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-réseau", multi=True, properties=(disabled_property,))
|
|
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
|
|
dyn = DynOptionDescription('leader', '', [interface1], suffixes=Calculation(return_list))
|
|
od = OptionDescription('toto', '', [dyn])
|
|
async with await Config(od) as cfg:
|
|
await cfg.property.read_write()
|
|
assert await cfg.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.get() == []
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.set(['192.168.1.2'])
|
|
assert await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 0).value.get() is None
|
|
assert await cfg.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.get() == ['192.168.1.2']
|
|
#
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.set(['192.168.1.2', '192.168.1.1'])
|
|
assert await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 0).value.get() is None
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 1).value.get()
|
|
#
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.set(['192.168.1.2', '192.168.1.2'])
|
|
assert await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 0).value.get() is None
|
|
assert await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 1).value.get() is None
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 1).value.set('255.255.255.255')
|
|
assert await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 1).value.get() == '255.255.255.255'
|
|
assert await cfg.value.dict() == {'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': ['192.168.1.2', '192.168.1.2'],
|
|
'leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1': [None, '255.255.255.255'],
|
|
'leaderval2.ip_admin_eth0val2.ip_admin_eth0val2': [],
|
|
'leaderval2.ip_admin_eth0val2.netmask_admin_eth0val2': []}
|
|
|
|
ret = await cfg.value.dict(leader_to_list=True)
|
|
assert ret == {'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': [{'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': '192.168.1.2', 'leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1': None},
|
|
{'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': '192.168.1.2', 'leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1': '255.255.255.255'}],
|
|
'leaderval2.ip_admin_eth0val2.ip_admin_eth0val2': []}
|
|
|
|
#
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.ip_admin_eth0val1').value.set(['192.168.1.2', '192.168.1.1'])
|
|
assert await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 0).value.get() is None
|
|
with pytest.raises(PropertiesOptionError):
|
|
await cfg.option('leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 1).value.get()
|
|
ret = await cfg.value.dict()
|
|
assert set(ret.keys()) == set(['leaderval1.ip_admin_eth0val1.ip_admin_eth0val1', 'leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1', 'leaderval2.ip_admin_eth0val2.ip_admin_eth0val2', 'leaderval2.ip_admin_eth0val2.netmask_admin_eth0val2'])
|
|
assert ret['leaderval1.ip_admin_eth0val1.ip_admin_eth0val1'] == ['192.168.1.2', '192.168.1.1']
|
|
assert len(ret['leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1']) == 2
|
|
assert ret['leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1'][0] is None
|
|
assert isinstance(ret['leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1'][1], PropertiesOptionError)
|
|
del ret['leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1'][1]
|
|
del ret['leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1'][0]
|
|
del ret['leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1']
|
|
assert await cfg.value.dict(leader_to_list=True) == {'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': [{'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': '192.168.1.2',
|
|
'leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1': None},
|
|
{'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': '192.168.1.1'}],
|
|
'leaderval2.ip_admin_eth0val2.ip_admin_eth0val2': []}
|
|
#
|
|
assert await cfg.value.dict(leader_to_list=True) == {'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': [{'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': '192.168.1.2',
|
|
'leaderval1.ip_admin_eth0val1.netmask_admin_eth0val1': None},
|
|
{'leaderval1.ip_admin_eth0val1.ip_admin_eth0val1': '192.168.1.1'}],
|
|
'leaderval2.ip_admin_eth0val2.ip_admin_eth0val2': []}
|
|
assert not await list_sessions()
|