# coding: utf-8 from time import sleep, time from py.test import raises from .autopath import do_autopath do_autopath() from tiramisu import setting, value setting.expires_time = 1 value.expires_time = 1 from tiramisu.option import BoolOption, IPOption, IntOption, StrOption, OptionDescription, MasterSlaves from tiramisu import Config from tiramisu.error import ConfigError, PropertiesOptionError from tiramisu.setting import groups from tiramisu import undefined, Params, ParamValue, ParamOption from tiramisu.api import TIRAMISU_VERSION from tiramisu.storage import list_sessions def teardown_function(function): 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 return_value(val): return val 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() assert od1.impl_already_build_caches() is False c = Config(od1) assert od1.impl_already_build_caches() is True c def test_cache(): od1 = make_description() cfg = Config(od1) values = cfg._config_bag.context.cfgimpl_get_values() settings = cfg._config_bag.context.cfgimpl_get_settings() cfg.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() cfg.option('u2').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u2' in values._p_.get_cached() assert 'u2' in settings._p_.get_cached() def test_cache_importation(): 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': []} def test_cache_importation_property(): 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'} def test_cache_importation_permissive(): 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'} def test_cache_reset(): od1 = make_description() cfg = Config(od1) values = cfg._config_bag.context.cfgimpl_get_values() settings = cfg._config_bag.context.cfgimpl_get_settings() #when change a value cfg.option('u1').value.get() cfg.option('u2').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u2' in values._p_.get_cached() assert 'u2' in settings._p_.get_cached() assert 'u1' in values._p_.get_cached() cfg.option('u2').value.set(1) assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u2' not in values._p_.get_cached() assert 'u2' not in settings._p_.get_cached() #when remove a value cfg.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() cfg.option('u2').value.reset() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u2' not in values._p_.get_cached() assert 'u2' not in settings._p_.get_cached() #when add/del property cfg.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() cfg.option('u2').property.add('test') assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u2' not in values._p_.get_cached() assert 'u2' not in settings._p_.get_cached() cfg.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() cfg.option('u2').property.pop('test') assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u2' not in values._p_.get_cached() assert 'u2' not in settings._p_.get_cached() #when enable/disabled property cfg.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() cfg.property.add('test') assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() cfg.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() cfg.property.pop('test') assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() def test_cache_reset_multi(): od1 = make_description() cfg = Config(od1) values = cfg._config_bag.context.cfgimpl_get_values() settings = cfg._config_bag.context.cfgimpl_get_settings() cfg.option('u1').value.get() cfg.option('u3').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u3' in values._p_.get_cached() assert 'u3' in settings._p_.get_cached() #when change a value cfg.option('u3').value.set([1]) assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u3' not in values._p_.get_cached() assert 'u3' not in settings._p_.get_cached() #when append value cfg.option('u1').value.get() cfg.option('u3').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u3' in values._p_.get_cached() assert 'u3' in settings._p_.get_cached() cfg.option('u3').value.set([1, 1]) assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u3' not in values._p_.get_cached() assert 'u3' not in settings._p_.get_cached() #when pop value cfg.option('u1').value.get() cfg.option('u3').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u3' in values._p_.get_cached() assert 'u3' in settings._p_.get_cached() cfg.option('u3').value.set([1]) assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u3' not in values._p_.get_cached() assert 'u3' not in settings._p_.get_cached() #when remove a value cfg.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() cfg.option('u3').value.reset() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u3' not in values._p_.get_cached() assert 'u3' not in settings._p_.get_cached() def test_reset_cache(): od1 = make_description() cfg = Config(od1) values = cfg._config_bag.context.cfgimpl_get_values() settings = cfg._config_bag.context.cfgimpl_get_settings() cfg.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() cfg.cache.reset() assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() cfg.option('u1').value.get() sleep(1) cfg.option('u1').value.get() sleep(1) cfg.option('u2').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u2' in values._p_.get_cached() assert 'u2' in settings._p_.get_cached() cfg.cache.reset() assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() assert 'u2' not in values._p_.get_cached() assert 'u2' not in settings._p_.get_cached() def test_cache_not_cache(): od1 = make_description() cfg = Config(od1) values = cfg._config_bag.context.cfgimpl_get_values() settings = cfg._config_bag.context.cfgimpl_get_settings() cfg.property.pop('cache') cfg.option('u1').value.get() assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() def test_cache_master_slave(): 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 = MasterSlaves('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0]) #interface1.impl_set_group_type(groups.master) maconfig = OptionDescription('toto', '', [interface1]) cfg = Config(maconfig) cfg.property.read_write() assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.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 = cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() if TIRAMISU_VERSION == 2: assert set(cache.keys()) == set(['ip_admin_eth0.ip_admin_eth0']) else: 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 = cfg._config_bag.context.cfgimpl_get_settings()._p_.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]) if TIRAMISU_VERSION == 2: assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([None, 0]) else: assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([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 = cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() if TIRAMISU_VERSION == 2: assert set(cache.keys()) == set(['ip_admin_eth0.ip_admin_eth0']) else: 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 = cfg._config_bag.context.cfgimpl_get_settings()._p_.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]) if TIRAMISU_VERSION == 2: assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([None, 0, 1]) else: assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([0, 1]) #DEL, insert, ... def return_value(value=None): return value 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(): val1 = StrOption('val1', "", 'val') val2 = StrOption('val2', "", callback=return_value, callback_params=Params((ParamOption(val1),)), properties=('mandatory',)) val3 = StrOption('val3', "", callback=return_value, callback_params=Params((ParamValue('yes'),))) val4 = StrOption('val4', "", callback=return_value, callback_params=Params(kwargs={'value': ParamOption(val1)})) val5 = StrOption('val5', "", callback=return_value, callback_params=Params(kwargs={'value': ParamValue('yes')}), multi=True) maconfig = OptionDescription('rootconfig', '', [val1, val2, val3, val4, val5]) cfg = Config(maconfig) cfg.property.read_write() if TIRAMISU_VERSION == 2: cfg.property.pop('expire') cfg.value.dict() #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.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') #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val3': {None: ('yes', None)}, 'val5': {None: (['yes'], None)}}) cfg.value.dict() #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.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') #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}, # 'val3': {None: (set([]), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val1': {None: ('new', None)}, 'val2': {None: ('new', None)}, 'val4': {None: ('new', None)}, 'val5': {None: (['yes'], None)}}) cfg.value.dict() #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}, # 'val3': {None: (set([]), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.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') #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}, # 'val3': {None: (set([]), None)}, # 'val4': {None: (set([]), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val1': {None: ('new', None)}, 'val2': {None: ('new', None)}, 'val3': {None: ('new2', None)}, 'val5': {None: (['yes'], None)}}) cfg.value.dict() #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}, # 'val3': {None: (set([]), None)}, # 'val4': {None: (set([]), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val1': {None: ('new', None)}, 'val2': {None: ('new', None)}, 'val3': {None: ('new2', None)}, 'val4': {None: ('new3', None)}, 'val5': {None: (['yes'], None)}}) cfg.option('val5').value.set([undefined, 'new4']) #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}, # 'val3': {None: (set([]), None)}, # 'val4': {None: (set([]), None)}, # 'val5': {None: (set(['empty']), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val1': {None: ('new', None)}, 'val2': {None: ('new', None)}, 'val3': {None: ('new2', None)}, 'val4': {None: ('new3', None)}}) cfg.value.dict() #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}, # 'val3': {None: (set([]), None)}, # 'val4': {None: (set([]), None)}, # 'val5': {None: (set(['empty']), None)}} compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val1': {None: ('new', None)}, 'val2': {None: ('new', None)}, 'val3': {None: ('new2', None)}, 'val4': {None: ('new3', None)}, 'val5': {None: (['yes', 'new4'], None)}}) def test_cache_master_and_slaves_master(): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", multi=True) interface1 = MasterSlaves('val1', '', [val1, val2]) #interface1.impl_set_group_type(groups.master) maconfig = OptionDescription('rootconfig', '', [interface1]) cfg = Config(maconfig) cfg.property.read_write() cfg.value.dict() global_props = ['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value'] val1_props = [] val1_val1_props = ['empty'] val1_val2_props = [] if TIRAMISU_VERSION == 2: cfg.property.pop('expire') global_props = set(global_props) val1_props = set(val1_props) val1_val1_props = set(val1_val1_props) val1_val2_props = set(val1_val2_props) else: 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 if TIRAMISU_VERSION == 2: assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {} assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} else: compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val1.val1': {None: ([], None)}}) # cfg.option('val1.val1').value.set([undefined]) compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached(), {None: {None: (set(global_props), None)}}) assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} cfg.value.dict() if TIRAMISU_VERSION == 2: val_val2 = [None] val_val2_props = {None: (set(), None), 0: (set(), None)} else: #has value idx_val2 = 0 val_val2 = None val_val2_props = {idx_val2: (val1_val2_props, None), None: (set(), None)} compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val1.val1': {None: ([None], None)}, 'val1.val2': {idx_val2: (val_val2, None)}}) cfg.option('val1.val1').value.set([undefined, undefined]) cfg.value.dict() cfg.option('val1.val2', 1).value.set('oui') compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)}}) assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} if TIRAMISU_VERSION == 2: val1_val2_props = {None: (set([]), None), 0: (set([]), None), 1: (set([]), None)} else: val1_val2_props = {0: (frozenset([]), None), 1: (frozenset([]), None)} #assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (global_props, None)}, # 'val1': {None: (val1_props, None)}, # 'val1.val1': {None: (val1_val1_props, None)}, # 'val1.val2': val1_val2_props} #if TIRAMISU_VERSION == 2: # assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None, None], None)}, # 'val1.val2': {None: ([None, 'oui'], None)}} #else: # assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None, None], None)}, # 'val1.val2': {0: (None, None), 1: ('oui', None)}} def test_cache_master_callback(): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", multi=True, callback=return_value, callback_params=Params(kwargs={'value': ParamOption(val1)})) interface1 = MasterSlaves('val1', '', [val1, val2]) #interface1.impl_set_group_type(groups.master) maconfig = OptionDescription('rootconfig', '', [interface1]) cfg = Config(maconfig) cfg.property.read_write() cfg.value.dict() global_props = ['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value'] val1_props = [] val1_val1_props = ['empty'] val1_val2_props = [] if TIRAMISU_VERSION == 2: cfg.property.pop('expire') global_props = set(global_props) val1_props = set(val1_props) val1_val1_props = set(val1_val1_props) val1_val2_props = set(val1_val2_props) else: global_props = frozenset(global_props) val1_props = frozenset(val1_props) val1_val1_props = frozenset(val1_val1_props) val1_val2_props = frozenset(val1_val2_props) if TIRAMISU_VERSION == 2: assert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {} assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} else: compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'val1.val1': {None: ([], None)}}) cfg.option('val1.val1').value.set([undefined]) compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached(), {None: {None: (set(global_props), None)}}) assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} cfg.value.dict() #FIXMEassert cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}} #FIXMEassert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None], None)}, # 'val1.val2': {None: ([None], None)} # } def test_cache_requires(): a = BoolOption('activate_service', '', True) b = IPOption('ip_address_service', '', requires=[{'option': a, 'expected': False, 'action': 'disabled'}]) od = OptionDescription('service', '', [a, b]) cfg = Config(od) cfg.property.read_write() if TIRAMISU_VERSION == 2: cfg.property.pop('expire') assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} assert cfg.option('ip_address_service').value.get() == None compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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)}}) if TIRAMISU_VERSION == 2: assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}} else: compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'ip_address_service': {None: (None, None)}, 'activate_service': {None: (True, None)}}) cfg.value.dict() compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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(cfg._config_bag.context.cfgimpl_get_values()._p_.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(cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)}, 'activate_service': {None: (set([]), None)}}) compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'activate_service': {None: (True, None)}}) cfg.value.dict() compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'ip_address_service': {None: ('1.1.1.1', None)}, 'activate_service': {None: (True, None)}}) cfg.option('activate_service').value.set(False) compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.get_cached(), {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings', 'force_store_value']), None)}}) assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} cfg.value.dict() compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'activate_service': {None: (False, None)}}) def test_cache_global_properties(): a = BoolOption('activate_service', '', True) b = IPOption('ip_address_service', '', requires=[{'option': a, 'expected': False, 'action': 'disabled'}]) od = OptionDescription('service', '', [a, b]) cfg = Config(od) cfg.property.read_write() assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {} assert cfg.option('ip_address_service').value.get() == None compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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)}}) if TIRAMISU_VERSION == 2: assert cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}} else: compare(cfg._config_bag.context.cfgimpl_get_values()._p_.get_cached(), {'ip_address_service': {None: (None, None)}, 'activate_service': {None: (True, None)}}) cfg.property.pop('disabled') assert cfg.option('ip_address_service').value.get() == None compare(cfg._config_bag.context.cfgimpl_get_settings()._p_.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(cfg._config_bag.context.cfgimpl_get_settings()._p_.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)}}) def test_callback_value_incr(): val1 = IntOption('val1', "", callback=return_incr, properties=('expire',)) val2 = IntOption('val2', "", callback=return_value, callback_params=Params(kwargs={'value': ParamOption(val1)})) maconfig = OptionDescription('rootconfig', '', [val1, val2]) cfg = Config(maconfig) 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