577 lines
28 KiB
Python
577 lines
28 KiB
Python
# coding: utf-8
|
|
|
|
from time import sleep, time
|
|
import pytest
|
|
from .autopath import do_autopath
|
|
do_autopath()
|
|
|
|
from tiramisu import BoolOption, IPOption, IntOption, StrOption, OptionDescription, Leadership, Config, \
|
|
undefined, Calculation, Params, ParamValue, ParamOption, \
|
|
list_sessions, default_storage, delete_session, calc_value
|
|
from tiramisu.error import ConfigError, PropertiesOptionError
|
|
from tiramisu.setting import groups
|
|
|
|
|
|
def teardown_function(function):
|
|
if default_storage.is_persistent():
|
|
sessions = list_sessions()
|
|
if not sessions:
|
|
return
|
|
assert len(sessions) == 1
|
|
delete_session(sessions[0])
|
|
else:
|
|
assert list_sessions() == [], 'session list is not empty when leaving "{}"'.format(function.__name__)
|
|
|
|
|
|
global incr
|
|
incr = -1
|
|
def return_incr():
|
|
global incr
|
|
incr += 1
|
|
return int(incr/2) + 1
|
|
|
|
|
|
def make_description():
|
|
u1 = IntOption('u1', '', multi=True)
|
|
u2 = IntOption('u2', '')
|
|
u3 = IntOption('u3', '', multi=True)
|
|
return OptionDescription('od1', '', [u1, u2, u3])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_config():
|
|
od1 = make_description()
|
|
assert od1.impl_already_build_caches() is False
|
|
c = await Config(od1)
|
|
assert od1.impl_already_build_caches() is True
|
|
c
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache():
|
|
od1 = make_description()
|
|
cfg = await Config(od1)
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
await cfg.option('u2').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u2' in values.get_cached()
|
|
assert 'u2' in settings.get_cached()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_importation():
|
|
od1 = make_description()
|
|
cfg = await Config(od1)
|
|
await cfg.option('u2').value.set(1)
|
|
export = await cfg.value.exportation()
|
|
assert await cfg.value.dict() == {'u1': [], 'u2': 1, 'u3': []}
|
|
await cfg.option('u2').value.set(2)
|
|
assert await cfg.value.dict() == {'u1': [], 'u2': 2, 'u3': []}
|
|
await cfg.value.importation(export)
|
|
assert await cfg.value.dict() == {'u1': [], 'u2': 1, 'u3': []}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_importation_property():
|
|
od1 = make_description()
|
|
cfg = await Config(od1)
|
|
await cfg.option('u2').property.add('prop')
|
|
export = await cfg.property.exportation()
|
|
assert await cfg.option('u2').property.get() == {'prop'}
|
|
await cfg.option('u2').property.add('prop2')
|
|
assert await cfg.option('u2').property.get() == {'prop', 'prop2'}
|
|
await cfg.property.importation(export)
|
|
assert await cfg.option('u2').property.get() == {'prop'}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_importation_permissive():
|
|
od1 = make_description()
|
|
cfg = await Config(od1)
|
|
await cfg.option('u2').permissive.set(frozenset(['prop']))
|
|
export = await cfg.permissive.exportation()
|
|
assert await cfg.option('u2').permissive.get() == {'prop'}
|
|
await cfg.option('u2').permissive.set(frozenset(['prop', 'prop2']))
|
|
assert await cfg.option('u2').permissive.get() == {'prop', 'prop2'}
|
|
await cfg.permissive.importation(export)
|
|
assert await cfg.option('u2').permissive.get() == {'prop'}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_reset():
|
|
od1 = make_description()
|
|
cfg = await Config(od1)
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
#when change a value
|
|
await cfg.option('u1').value.get()
|
|
await cfg.option('u2').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u2' in values.get_cached()
|
|
assert 'u2' in settings.get_cached()
|
|
assert 'u1' in values.get_cached()
|
|
settings.get_cached()
|
|
await cfg.option('u2').value.set(1)
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u2' in values.get_cached()
|
|
assert 'u2' not in settings.get_cached()
|
|
#when remove a value
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
await cfg.option('u2').value.reset()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u2' not in values.get_cached()
|
|
assert 'u2' not in settings.get_cached()
|
|
#when add/del property
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
await cfg.option('u2').property.add('test')
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u2' not in values.get_cached()
|
|
assert 'u2' not in settings.get_cached()
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
await cfg.option('u2').property.pop('test')
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u2' not in values.get_cached()
|
|
assert 'u2' not in settings.get_cached()
|
|
#when enable/disabled property
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
await cfg.property.add('test')
|
|
assert 'u1' not in values.get_cached()
|
|
assert 'u1' not in settings.get_cached()
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
await cfg.property.pop('test')
|
|
assert 'u1' not in values.get_cached()
|
|
assert 'u1' not in settings.get_cached()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_reset_multi():
|
|
od1 = make_description()
|
|
cfg = await Config(od1)
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
await cfg.option('u1').value.get()
|
|
await cfg.option('u3').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u3' in values.get_cached()
|
|
assert 'u3' in settings.get_cached()
|
|
#when change a value
|
|
await cfg.option('u3').value.set([1])
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u3' in values.get_cached()
|
|
assert 'u3' not in settings.get_cached()
|
|
#when append value
|
|
await cfg.option('u1').value.get()
|
|
await cfg.option('u3').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u3' in values.get_cached()
|
|
assert 'u3' in settings.get_cached()
|
|
await cfg.option('u3').value.set([1, 2])
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u3' in values.get_cached()
|
|
assert 'u3' not in settings.get_cached()
|
|
#when pop value
|
|
await cfg.option('u1').value.get()
|
|
await cfg.option('u3').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u3' in values.get_cached()
|
|
assert 'u3' in settings.get_cached()
|
|
await cfg.option('u3').value.set([1])
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u3' in values.get_cached()
|
|
assert 'u3' not in settings.get_cached()
|
|
#when remove a value
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
await cfg.option('u3').value.reset()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u3' not in values.get_cached()
|
|
assert 'u3' not in settings.get_cached()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_cache():
|
|
od1 = make_description()
|
|
cfg = await Config(od1)
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
await cfg.cache.reset()
|
|
assert 'u1' not in values.get_cached()
|
|
assert 'u1' not in settings.get_cached()
|
|
await cfg.option('u1').value.get()
|
|
await cfg.option('u2').value.get()
|
|
assert 'u1' in values.get_cached()
|
|
assert 'u1' in settings.get_cached()
|
|
assert 'u2' in values.get_cached()
|
|
assert 'u2' in settings.get_cached()
|
|
await cfg.cache.reset()
|
|
assert 'u1' not in values.get_cached()
|
|
assert 'u1' not in settings.get_cached()
|
|
assert 'u2' not in values.get_cached()
|
|
assert 'u2' not in settings.get_cached()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_not_cache():
|
|
od1 = make_description()
|
|
cfg = await Config(od1)
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
await cfg.property.pop('cache')
|
|
await cfg.option('u1').value.get()
|
|
assert 'u1' not in values.get_cached()
|
|
assert 'u1' not in settings.get_cached()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_leadership():
|
|
ip_admin_eth0 = StrOption('ip_admin_eth0', "ip réseau autorisé", multi=True)
|
|
netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-réseau", multi=True)
|
|
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
|
|
od1 = OptionDescription('toto', '', [interface1])
|
|
cfg = await Config(od1)
|
|
await cfg.property.read_write()
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
assert values.get_cached() == {}
|
|
#assert settings.get_cached() == {}
|
|
#
|
|
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['192.168.1.2'])
|
|
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get()
|
|
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.get()
|
|
cache = values.get_cached()
|
|
assert set(cache.keys()) == set(['ip_admin_eth0.ip_admin_eth0', 'ip_admin_eth0.netmask_admin_eth0'])
|
|
assert set(cache['ip_admin_eth0.ip_admin_eth0'].keys()) == set([None])
|
|
assert cache['ip_admin_eth0.ip_admin_eth0'][None][0] == ['192.168.1.2']
|
|
#assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([None])
|
|
#assert cache['ip_admin_eth0.netmask_admin_eth0'][None][0] == [None]
|
|
#assert cache['ip_admin_eth0.netmask_admin_eth0'][0][0] is None
|
|
cache = settings.get_cached()
|
|
assert set(cache.keys()) == set([None, 'ip_admin_eth0', 'ip_admin_eth0.ip_admin_eth0', 'ip_admin_eth0.netmask_admin_eth0'])
|
|
assert set(cache['ip_admin_eth0'].keys()) == set([None])
|
|
assert set(cache['ip_admin_eth0.ip_admin_eth0'].keys()) == set([None])
|
|
assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([0, None])
|
|
#
|
|
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['192.168.1.2', '192.168.1.1'])
|
|
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get()
|
|
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.get()
|
|
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.get()
|
|
cache = values.get_cached()
|
|
assert set(cache.keys()) == set(['ip_admin_eth0.ip_admin_eth0', 'ip_admin_eth0.netmask_admin_eth0'])
|
|
assert set(cache['ip_admin_eth0.ip_admin_eth0'].keys()) == set([None])
|
|
assert cache['ip_admin_eth0.ip_admin_eth0'][None][0] == ['192.168.1.2', '192.168.1.1']
|
|
#assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([None])
|
|
#assert cache['ip_admin_eth0.netmask_admin_eth0'][None][0] == [None, None]
|
|
#assert cache['ip_admin_eth0.netmask_admin_eth0'][0][0] is None
|
|
#assert cache['ip_admin_eth0.netmask_admin_eth0'][1][0] is None
|
|
cache = settings.get_cached()
|
|
assert set(cache.keys()) == set([None, 'ip_admin_eth0', 'ip_admin_eth0.ip_admin_eth0', 'ip_admin_eth0.netmask_admin_eth0'])
|
|
assert set(cache['ip_admin_eth0'].keys()) == set([None])
|
|
assert set(cache['ip_admin_eth0.ip_admin_eth0'].keys()) == set([None])
|
|
assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([None, 0, 1])
|
|
#DEL, insert, ...
|
|
|
|
|
|
def compare(calculated, expected):
|
|
assert set(calculated.keys()) == set(expected.keys())
|
|
for calculated_key in calculated:
|
|
assert set(calculated[calculated_key].keys()) == set(expected[calculated_key].keys())
|
|
for calculated_subkey in calculated[calculated_key]:
|
|
# do not check timestamp
|
|
assert calculated[calculated_key][calculated_subkey][0] == expected[calculated_key][calculated_subkey][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_callback():
|
|
val1 = StrOption('val1', "", 'val')
|
|
val2 = StrOption('val2', "", Calculation(calc_value, Params(ParamOption(val1))), properties=('mandatory',))
|
|
val3 = StrOption('val3', "", Calculation(calc_value, Params(ParamValue('yes'))))
|
|
val4 = StrOption('val4', "", Calculation(calc_value, Params(ParamOption(val1))))
|
|
val5 = StrOption('val5', "", [Calculation(calc_value, Params(ParamValue('yes')))], multi=True)
|
|
od1 = OptionDescription('rootconfig', '', [val1, val2, val3, val4, val5])
|
|
cfg = await Config(od1)
|
|
await cfg.property.read_write()
|
|
await cfg.value.dict()
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
compare(values.get_cached(), {'val1': {None: ('val', None)},
|
|
'val2': {None: ('val', None)},
|
|
'val3': {None: ('yes', None)},
|
|
'val4': {None: ('val', None)},
|
|
'val5': {None: (['yes'], None)}})
|
|
await cfg.option('val1').value.set('new')
|
|
compare(values.get_cached(), {'val3': {None: ('yes', None)},
|
|
'val1': {None: ('new', None)},
|
|
'val5': {None: (['yes'], None)}})
|
|
await cfg.value.dict()
|
|
compare(values.get_cached(), {'val1': {None: ('new', None)},
|
|
'val2': {None: ('new', None)},
|
|
'val3': {None: ('yes', None)},
|
|
'val4': {None: ('new', None)},
|
|
'val5': {None: (['yes'], None)}})
|
|
await cfg.option('val3').value.set('new2')
|
|
compare(values.get_cached(), {'val1': {None: ('new', None)},
|
|
'val2': {None: ('new', None)},
|
|
'val4': {None: ('new', None)},
|
|
'val1': {None: ('new', None)},
|
|
'val3': {None: ('new2', None, True)},
|
|
'val5': {None: (['yes'], None)}})
|
|
await cfg.value.dict()
|
|
compare(values.get_cached(), {'val1': {None: ('new', None)},
|
|
'val2': {None: ('new', None)},
|
|
'val3': {None: ('new2', None)},
|
|
'val4': {None: ('new', None)},
|
|
'val5': {None: (['yes'], None)}})
|
|
await cfg.option('val4').value.set('new3')
|
|
compare(values.get_cached(), {'val1': {None: ('new', None)},
|
|
'val2': {None: ('new', None)},
|
|
'val3': {None: ('new2', None)},
|
|
'val4': {None: ('new3', None, True)},
|
|
'val5': {None: (['yes'], None)}})
|
|
await cfg.value.dict()
|
|
compare(values.get_cached(), {'val1': {None: ('new', None)},
|
|
'val2': {None: ('new', None)},
|
|
'val3': {None: ('new2', None)},
|
|
'val4': {None: ('new3', None)},
|
|
'val5': {None: (['yes'], None)}})
|
|
await cfg.option('val5').value.set([undefined, 'new4'])
|
|
compare(values.get_cached(), {'val1': {None: ('new', None)},
|
|
'val2': {None: ('new', None)},
|
|
'val3': {None: ('new2', None)},
|
|
'val4': {None: ('new3', None)},
|
|
'val5': {None: (['yes', 'new4'], None)}})
|
|
await cfg.value.dict()
|
|
compare(values.get_cached(), {'val1': {None: ('new', None)},
|
|
'val2': {None: ('new', None)},
|
|
'val3': {None: ('new2', None)},
|
|
'val4': {None: ('new3', None)},
|
|
'val5': {None: (['yes', 'new4'], None)}})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_leader_and_followers():
|
|
val1 = StrOption('val1', "", multi=True)
|
|
val2 = StrOption('val2', "", multi=True)
|
|
interface1 = Leadership('val1', '', [val1, val2])
|
|
od1 = OptionDescription('rootconfig', '', [interface1])
|
|
cfg = await Config(od1)
|
|
await cfg.property.read_write()
|
|
await cfg.value.dict()
|
|
global_props = ['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']
|
|
val1_props = []
|
|
val1_val1_props = ['empty', 'unique']
|
|
val1_val2_props = []
|
|
global_props = frozenset(global_props)
|
|
val1_props = frozenset(val1_props)
|
|
val1_val1_props = frozenset(val1_val1_props)
|
|
val1_val2_props = frozenset(val1_val2_props)
|
|
#None because no value
|
|
idx_val2 = None
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
compare(settings.get_cached(), {None: {None: (global_props, None)},
|
|
'val1': {None: (val1_props, None)},
|
|
'val1.val1': {None: (val1_val1_props, None)},
|
|
'val1.val2': {idx_val2: (val1_val2_props, None)}})
|
|
# len is 0 so don't get any value
|
|
compare(values.get_cached(), {'val1.val1': {None: ([], None)}})
|
|
#
|
|
await cfg.option('val1.val1').value.set([undefined])
|
|
val_val2_props = {idx_val2: (val1_val2_props, None), None: (set(), None)}
|
|
compare(settings.get_cached(), {None: {None: (set(global_props), None)},
|
|
'val1.val1': {None: (val1_val1_props, None)},
|
|
'val1.val2': val_val2_props})
|
|
compare(values.get_cached(), {'val1.val1': {None: ([None], None, True)}})
|
|
await cfg.value.dict()
|
|
#has value
|
|
idx_val2 = 0
|
|
val_val2 = None
|
|
val_val2_props = {idx_val2: (val1_val2_props, None), None: (set(), None)}
|
|
compare(settings.get_cached(), {None: {None: (global_props, None)},
|
|
'val1': {None: (val1_props, None)},
|
|
'val1.val1': {None: (val1_val1_props, None)},
|
|
'val1.val2': val_val2_props})
|
|
compare(values.get_cached(), {'val1.val1': {None: ([None], None)},
|
|
'val1.val2': {idx_val2: (val_val2, None)}})
|
|
await cfg.option('val1.val1').value.set([undefined, undefined])
|
|
await cfg.value.dict()
|
|
await cfg.option('val1.val2', 1).value.set('oui')
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)}})
|
|
compare(values.get_cached(), {'val1.val2': {1: ('oui', None, True)}})
|
|
val1_val2_props = {0: (frozenset([]), None), 1: (frozenset([]), None)}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_leader_callback():
|
|
val1 = StrOption('val1', "", multi=True)
|
|
val2 = StrOption('val2', "", Calculation(calc_value, Params(kwargs={'value': ParamOption(val1)})), multi=True)
|
|
interface1 = Leadership('val1', '', [val1, val2])
|
|
od1 = OptionDescription('rootconfig', '', [interface1])
|
|
cfg = await Config(od1)
|
|
await cfg.property.read_write()
|
|
await cfg.value.dict()
|
|
global_props = ['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']
|
|
val1_props = []
|
|
val1_val1_props = ['empty', 'unique']
|
|
val1_val2_props = []
|
|
global_props = frozenset(global_props)
|
|
val1_props = frozenset(val1_props)
|
|
val1_val1_props = frozenset(val1_val1_props)
|
|
val1_val2_props = frozenset(val1_val2_props)
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
compare(settings.get_cached(), {None: {None: (global_props, None)},
|
|
'val1': {None: (val1_props, None)},
|
|
'val1.val1': {None: (val1_val1_props, None)},
|
|
'val1.val2': {None: (val1_val2_props, None)}})
|
|
compare(values.get_cached(), {'val1.val1': {None: ([], None)}})
|
|
await cfg.option('val1.val1').value.set([undefined])
|
|
compare(settings.get_cached(), {None: {None: (set(global_props), None)},
|
|
'val1': {None: (set(), None, True)},
|
|
'val1.val1': {None: (val1_val1_props, None)},
|
|
'val1.val2': {None: (val1_val2_props, None)}})
|
|
|
|
compare(values.get_cached(), {'val1.val1': {None: ([None], None, True)}})
|
|
await cfg.value.dict()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_requires():
|
|
a = BoolOption('activate_service', '', True)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(a),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
b = IPOption('ip_address_service', '', properties=(disabled_property,))
|
|
od = OptionDescription('service', '', [a, b])
|
|
cfg = await Config(od)
|
|
await cfg.property.read_write()
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
assert values.get_cached() == {}
|
|
assert await cfg.option('ip_address_service').value.get() == None
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)},
|
|
'activate_service': {None: (set([]), None)},
|
|
'ip_address_service': {None: (set([]), None)}})
|
|
|
|
compare(values.get_cached(), {'ip_address_service': {None: (None, None)},
|
|
'activate_service': {None: (True, None)}})
|
|
await cfg.value.dict()
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)},
|
|
'activate_service': {None: (set([]), None)},
|
|
'ip_address_service': {None: (set([]), None)}})
|
|
|
|
compare(values.get_cached(), {'ip_address_service': {None: (None, None)},
|
|
'activate_service': {None: (True, None)}})
|
|
await cfg.option('ip_address_service').value.set('1.1.1.1')
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)},
|
|
'activate_service': {None: (set([]), None)}})
|
|
|
|
compare(values.get_cached(), {'activate_service': {None: (True, None)}, 'ip_address_service': {None: ('1.1.1.1', None, True)}})
|
|
await cfg.value.dict()
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)},
|
|
'activate_service': {None: (set([]), None)},
|
|
'ip_address_service': {None: (set([]), None)}})
|
|
|
|
compare(values.get_cached(), {'ip_address_service': {None: ('1.1.1.1', None)},
|
|
'activate_service': {None: (True, None)}})
|
|
await cfg.option('activate_service').value.set(False)
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)}})
|
|
|
|
compare(values.get_cached(), {'activate_service': {None: (False, None)}})
|
|
await cfg.value.dict()
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)},
|
|
'activate_service': {None: (set([]), None)},
|
|
'ip_address_service': {None: (set(['disabled']), None)}})
|
|
|
|
compare(values.get_cached(), {'activate_service': {None: (False, None)}})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_global_properties():
|
|
a = BoolOption('activate_service', '', True)
|
|
disabled_property = Calculation(calc_value,
|
|
Params(ParamValue('disabled'),
|
|
kwargs={'condition': ParamOption(a),
|
|
'expected': ParamValue(False),
|
|
'default': ParamValue(None)}))
|
|
b = IPOption('ip_address_service', '', properties=(disabled_property,))
|
|
od = OptionDescription('service', '', [a, b])
|
|
cfg = await Config(od)
|
|
await cfg.property.read_write()
|
|
values = cfg._config_bag.context._impl_values_cache
|
|
settings = cfg._config_bag.context._impl_properties_cache
|
|
assert values.get_cached() == {}
|
|
assert await cfg.option('ip_address_service').value.get() == None
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)},
|
|
'activate_service': {None: (set([]), None)},
|
|
'ip_address_service': {None: (set([]), None)}})
|
|
|
|
compare(values.get_cached(), {'ip_address_service': {None: (None, None)},
|
|
'activate_service': {None: (True, None)}})
|
|
await cfg.property.pop('disabled')
|
|
assert await cfg.option('ip_address_service').value.get() == None
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)},
|
|
'activate_service': {None: (set([]), None)},
|
|
'ip_address_service': {None: (set([]), None)}})
|
|
await cfg.property.add('test')
|
|
assert await cfg.option('ip_address_service').value.get() == None
|
|
compare(settings.get_cached(), {None: {None: (set(['cache', 'frozen', 'hidden', 'validator', 'warnings', 'test', 'force_store_value']), None)},
|
|
'activate_service': {None: (set([]), None)},
|
|
'ip_address_service': {None: (set([]), None)}})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_value_incr():
|
|
val1 = IntOption('val1', "", Calculation(return_incr), properties=('expire',))
|
|
val2 = IntOption('val2', "", Calculation(calc_value, Params(ParamOption(val1))))
|
|
od1 = OptionDescription('rootconfig', '', [val1, val2])
|
|
cfg = await Config(od1)
|
|
assert await cfg.cache.get_expiration_time() == 5
|
|
await cfg.cache.set_expiration_time(1)
|
|
assert await cfg.cache.get_expiration_time() == 1
|
|
await cfg.property.read_write()
|
|
assert await cfg.option('val1').value.get() == 1
|
|
sleep(1)
|
|
assert await cfg.option('val2').value.get() == 1
|
|
sleep(1)
|
|
assert await cfg.option('val1').value.get() == 1
|
|
assert await cfg.option('val2').value.get() == 1
|
|
sleep(2)
|
|
assert await cfg.option('val1').value.get() == 2
|
|
assert await cfg.option('val2').value.get() == 2
|
|
assert await cfg.option('val1').value.get() == 2
|
|
assert await cfg.option('val2').value.get() == 2
|