tiramisu/tests/test_cache.py

581 lines
26 KiB
Python
Raw Permalink Normal View History

# coding: utf-8
2018-10-31 08:00:19 +01:00
from time import sleep, time
2017-07-09 09:49:03 +02:00
from .autopath import do_autopath
2015-07-24 17:54:10 +02:00
do_autopath()
2019-09-01 09:41:53 +02:00
from tiramisu import BoolOption, IPOption, IntOption, StrOption, OptionDescription, Leadership, Config, \
undefined, Calculation, Params, ParamValue, ParamOption, calc_value
from tiramisu.error import ConfigError, PropertiesOptionError
from tiramisu.setting import groups
2017-09-17 15:55:32 +02:00
global incr
incr = -1
2017-09-17 15:55:32 +02:00
def return_incr():
global incr
incr += 1
return int(incr/2) + 1
2017-09-17 15:55:32 +02:00
def make_description():
u1 = IntOption('u1', '', multi=True)
u2 = IntOption('u2', '')
u3 = IntOption('u3', '', multi=True)
return OptionDescription('od1', '', [u1, u2, u3])
def test_cache_config():
od1 = make_description()
2015-07-24 17:54:10 +02:00
assert od1.impl_already_build_caches() is False
cfg = Config(od1)
assert od1.impl_already_build_caches() is True
cfg
# assert not list_sessions()
def test_cache():
od1 = make_description()
cfg = Config(od1)
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.properties_cache
cfg.option('u1').value.get()
assert 'u1' in values.get_cached()
assert 'u1' in settings.get_cached()
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 not list_sessions()
def test_cache_importation():
2018-04-27 14:20:12 +02:00
od1 = make_description()
cfg = Config(od1)
cfg.option('u2').value.set(1)
export = cfg.value.exportation()
assert cfg.value.dict() == {'u1': [], 'u2': 1, 'u3': []}
cfg.option('u2').value.set(2)
assert cfg.value.dict() == {'u1': [], 'u2': 2, 'u3': []}
cfg.value.importation(export)
assert cfg.value.dict() == {'u1': [], 'u2': 1, 'u3': []}
# assert not list_sessions()
def test_cache_importation_property():
2018-04-27 14:20:12 +02:00
od1 = make_description()
cfg = Config(od1)
cfg.option('u2').property.add('prop')
export = cfg.property.exportation()
assert cfg.option('u2').property.get() == {'prop'}
cfg.option('u2').property.add('prop2')
assert cfg.option('u2').property.get() == {'prop', 'prop2'}
cfg.property.importation(export)
assert cfg.option('u2').property.get() == {'prop'}
cfg = Config(od1)
# assert not list_sessions()
def test_cache_importation_permissive():
2018-04-27 14:20:12 +02:00
od1 = make_description()
cfg = Config(od1)
cfg.option('u2').permissive.set(frozenset(['prop']))
export = cfg.permissive.exportation()
assert cfg.option('u2').permissive.get() == {'prop'}
cfg.option('u2').permissive.set(frozenset(['prop', 'prop2']))
assert cfg.option('u2').permissive.get() == {'prop', 'prop2'}
cfg.permissive.importation(export)
assert cfg.option('u2').permissive.get() == {'prop'}
# assert not list_sessions()
def test_cache_reset():
od1 = make_description()
cfg = Config(od1)
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.properties_cache
#when change a value
cfg.option('u1').value.get()
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()
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
cfg.option('u1').value.get()
assert 'u1' in values.get_cached()
assert 'u1' in settings.get_cached()
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
cfg.option('u1').value.get()
assert 'u1' in values.get_cached()
assert 'u1' in settings.get_cached()
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()
cfg.option('u1').value.get()
assert 'u1' in values.get_cached()
assert 'u1' in settings.get_cached()
cfg.option('u2').property.remove('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
cfg.option('u1').value.get()
assert 'u1' in values.get_cached()
assert 'u1' in settings.get_cached()
cfg.property.add('test')
assert 'u1' not in values.get_cached()
assert 'u1' not in settings.get_cached()
cfg.option('u1').value.get()
assert 'u1' in values.get_cached()
assert 'u1' in values.get_cached()
assert 'u1' in settings.get_cached()
cfg.property.remove('test')
assert 'u1' not in values.get_cached()
assert 'u1' not in settings.get_cached()
# assert not list_sessions()
def test_cache_reset_multi():
od1 = make_description()
cfg = Config(od1)
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.properties_cache
cfg.option('u1').value.get()
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
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
cfg.option('u1').value.get()
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()
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
cfg.option('u1').value.get()
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()
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
cfg.option('u1').value.get()
assert 'u1' in values.get_cached()
assert 'u1' in settings.get_cached()
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()
# assert not list_sessions()
def test_reset_cache():
od1 = make_description()
cfg = Config(od1)
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.properties_cache
cfg.option('u1').value.get()
assert 'u1' in values.get_cached()
assert 'u1' in settings.get_cached()
cfg.cache.reset()
assert 'u1' not in values.get_cached()
assert 'u1' not in settings.get_cached()
cfg.option('u1').value.get()
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()
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()
# assert not list_sessions()
def test_cache_not_cache():
2013-09-07 17:25:22 +02:00
od1 = make_description()
cfg = Config(od1)
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.properties_cache
cfg.property.remove('cache')
cfg.option('u1').value.get()
assert 'u1' not in values.get_cached()
assert 'u1' not in settings.get_cached()
# assert not list_sessions()
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)
2019-02-23 19:06:23 +01:00
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
od1 = OptionDescription('toto', '', [interface1])
cfg = Config(od1)
cfg.property.read_write()
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.properties_cache
assert values.get_cached() == {}
#assert settings.get_cached() == {}
#
cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['192.168.1.2'])
cfg.option('ip_admin_eth0.ip_admin_eth0').value.get()
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])
2023-06-26 19:25:57 +02:00
assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == {0}
#
cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['192.168.1.2', '192.168.1.1'])
cfg.option('ip_admin_eth0.ip_admin_eth0').value.get()
cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.get()
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])
2023-06-26 19:25:57 +02:00
assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([0, 1])
#DEL, insert, ...
# assert not list_sessions()
2017-07-08 15:59:56 +02:00
2018-06-25 21:40:16 +02:00
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]
def test_cache_callback():
2017-07-08 15:59:56 +02:00
val1 = StrOption('val1', "", 'val')
2019-09-28 16:32:48 +02:00
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 = Config(od1)
cfg.property.read_write()
cfg.value.dict()
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.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)}})
cfg.option('val1').value.set('new')
compare(values.get_cached(), {'val3': {None: ('yes', None)},
'val1': {None: ('new', None)},
'val5': {None: (['yes'], None)}})
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)}})
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)}})
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)}})
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)}})
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)}})
# assert not list_sessions()
def test_cache_leader_and_followers():
2017-07-08 15:59:56 +02:00
val1 = StrOption('val1', "", multi=True)
val2 = StrOption('val2', "", multi=True)
2019-02-23 19:06:23 +01:00
interface1 = Leadership('val1', '', [val1, val2])
od1 = OptionDescription('rootconfig', '', [interface1])
cfg = Config(od1)
cfg.property.read_write()
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.properties_cache
compare(settings.get_cached(), {None: {None: (global_props, None)},
'val1': {None: (val1_props, None)},
'val1.val1': {None: (val1_val1_props, None)},
2023-04-27 11:34:35 +02:00
})
# len is 0 so don't get any value
compare(values.get_cached(), {'val1.val1': {None: ([], None)}})
#
cfg.option('val1.val1').value.set([None])
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)},
2023-06-26 19:25:57 +02:00
})
compare(values.get_cached(), {'val1.val1': {None: ([None], None, True)}})
cfg.value.dict()
#has value
idx_val2 = 0
val_val2 = None
2023-06-26 19:25:57 +02:00
val_val2_props = {idx_val2: (val1_val2_props, 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)},
2023-06-26 19:25:57 +02:00
'val1.val2': {idx_val2: (val_val2, None)},
})
cfg.option('val1.val1').value.set([None, None])
cfg.value.dict()
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)}
# assert not list_sessions()
def test_cache_leader_callback():
2017-07-08 15:59:56 +02:00
val1 = StrOption('val1', "", multi=True)
2019-09-28 16:32:48 +02:00
val2 = StrOption('val2', "", Calculation(calc_value, Params(kwargs={'value': ParamOption(val1)})), multi=True)
2019-02-23 19:06:23 +01:00
interface1 = Leadership('val1', '', [val1, val2])
od1 = OptionDescription('rootconfig', '', [interface1])
cfg = Config(od1)
cfg.property.read_write()
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.properties_cache
compare(settings.get_cached(), {None: {None: (global_props, None)},
'val1': {None: (val1_props, None)},
'val1.val1': {None: (val1_val1_props, None)},
2023-04-27 11:34:35 +02:00
})
compare(values.get_cached(), {'val1.val1': {None: ([], None)}})
cfg.option('val1.val1').value.set([None])
compare(settings.get_cached(), {None: {None: (set(global_props), None)},
'val1.val1': {None: (val1_val1_props, None)},
2023-06-26 19:25:57 +02:00
})
compare(values.get_cached(), {'val1.val1': {None: ([None], None, True)}})
cfg.value.dict()
# assert not list_sessions()
def test_cache_requires():
2017-07-08 15:59:56 +02:00
a = BoolOption('activate_service', '', True)
2019-09-01 09:41:53 +02:00
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,))
2020-01-22 20:46:18 +01:00
od1 = OptionDescription('service', '', [a, b])
cfg = Config(od1)
cfg.property.read_write()
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.properties_cache
assert values.get_cached() == {}
assert 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)}})
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)}})
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)}})
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)}})
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)}})
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)}})
# assert not list_sessions()
def test_cache_global_properties():
2017-07-13 22:04:06 +02:00
a = BoolOption('activate_service', '', True)
2019-09-01 09:41:53 +02:00
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,))
2020-01-22 20:46:18 +01:00
od1 = OptionDescription('service', '', [a, b])
cfg = Config(od1)
cfg.property.read_write()
values = cfg._config_bag.context._impl_values_cache
settings = cfg._config_bag.context.properties_cache
assert values.get_cached() == {}
assert 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)}})
cfg.property.remove('disabled')
assert 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)}})
cfg.property.add('test')
assert 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)}})
# assert not list_sessions()
def test_callback_value_incr():
global incr
incr = -1
val1 = IntOption('val1', "", Calculation(return_incr), properties=('expire',))
val2 = IntOption('val2', "", Calculation(calc_value, Params(ParamOption(val1))))
od1 = OptionDescription('rootconfig', '', [val1, val2])
cfg = Config(od1)
assert cfg.cache.get_expiration_time() == 5
cfg.cache.set_expiration_time(1)
assert cfg.cache.get_expiration_time() == 1
cfg.property.read_write()
assert cfg.option('val1').value.get() == 1
sleep(1)
assert cfg.option('val2').value.get() == 1
sleep(1)
assert cfg.option('val1').value.get() == 1
assert cfg.option('val2').value.get() == 1
sleep(2)
assert cfg.option('val1').value.get() == 2
assert cfg.option('val2').value.get() == 2
assert cfg.option('val1').value.get() == 2
assert cfg.option('val2').value.get() == 2
# assert not list_sessions()
def test_callback_value_incr_demoting():
global incr
incr = -1
2019-09-28 16:32:48 +02:00
val1 = IntOption('val1', "", Calculation(return_incr), properties=('expire',))
val2 = IntOption('val2', "", Calculation(calc_value, Params(ParamOption(val1))))
od1 = OptionDescription('rootconfig', '', [val1, val2])
cfg = Config(od1)
cfg.property.add('demoting_error_warning')
assert cfg.cache.get_expiration_time() == 5
cfg.cache.set_expiration_time(1)
assert cfg.cache.get_expiration_time() == 1
cfg.property.read_write()
assert cfg.option('val1').value.get() == 1
sleep(1)
assert cfg.option('val2').value.get() == 1
sleep(1)
assert cfg.option('val1').value.get() == 1
assert cfg.option('val2').value.get() == 1
sleep(2)
assert cfg.option('val1').value.get() == 2
assert cfg.option('val2').value.get() == 2
assert cfg.option('val1').value.get() == 2
assert cfg.option('val2').value.get() == 2
# assert not list_sessions()