# coding: utf-8 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.config import Config from tiramisu.error import ConfigError, PropertiesOptionError from tiramisu.setting import groups from tiramisu import getapi, undefined, Params, ParamValue, ParamOption from tiramisu.api import TIRAMISU_VERSION from time import sleep, time from py.test import raises 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() c = Config(od1) api = getapi(c) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() api.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() api.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_get_cache(): # # force a value in cache, try if reget corrupted value # od1 = make_description() # c = Config(od1) # api = getapi(c) # values = c.cfgimpl_get_values() # settings = c.cfgimpl_get_settings() # ntime = time() + 1 # settings._p_.setcache('u1', set(['inject']), ntime, None) # assert 'inject' in settings[od1.u1] # values._p_.setcache('u1', 100, ntime, None) # assert api.option('u1').value.get() == [100] #def test_get_cache_no_expire(): # # force a value in cache, try if reget corrupted value # od1 = make_description() # c = Config(od1) # api = getapi(c) # values = c.cfgimpl_get_values() # settings = c.cfgimpl_get_settings() # settings._p_.setcache('u1', set(['inject2']), None, None) # assert 'inject2' in settings[od1.u1] # values._p_.setcache('u1', 200, None, None) # assert api.option('u1').value.get() == [200] def test_cache_reset(): od1 = make_description() c = Config(od1) api = getapi(c) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() #when change a value api.option('u1').value.get() api.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() api.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 api.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() api.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 api.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() api.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() api.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() api.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 api.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() api.property.add('test') assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() api.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() api.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() c = Config(od1) api = getapi(c) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() api.option('u1').value.get() api.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 api.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 api.option('u1').value.get() api.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() api.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 api.option('u1').value.get() api.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() api.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 api.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() api.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() c = Config(od1) api = getapi(c) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() api.option('u1').value.get() assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() c.cfgimpl_reset_cache(None, None, None) assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() api.option('u1').value.get() sleep(1) api.option('u1').value.get() sleep(1) api.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() c.cfgimpl_reset_cache(None, None, None) 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_reset_cache_subconfig(): # od1 = make_description() # od2 = OptionDescription('od2', '', [od1]) # c = Config(od2) # api = getapi(c) # values = c.cfgimpl_get_values() # api.option('od1.u1').value.get() # assert 'od1.u1' in values._p_.get_cached() # c.od1.cfgimpl_reset_cache(None, None, None) # assert 'od1.u1' not in values._p_.get_cached() #def test_reset_cache_only_expired(): # od1 = make_description() # c = Config(od1) # api = getapi(c) # api.property.add('expire') # values = c.cfgimpl_get_values() # settings = c.cfgimpl_get_settings() # api.option('u1').value.get() # assert 'u1' in values._p_.get_cached() # assert 'u1' in settings._p_.get_cached() # c.cfgimpl_reset_cache(True) # assert 'u1' in values._p_.get_cached() # assert 'u1' in settings._p_.get_cached() # sleep(1) # api.option('u1').value.get() # sleep(1) # api.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() # c.cfgimpl_reset_cache(True) # assert 'u1' not in values._p_.get_cached() # assert 'u1' not in settings._p_.get_cached() # assert 'u2' in values._p_.get_cached() # assert 'u2' in settings._p_.get_cached() #def test_cache_not_expire(): # od1 = make_description() # c = Config(od1) # api = getapi(c) # values = c.cfgimpl_get_values() # settings = c.cfgimpl_get_settings() # #api.property.pop('expire') # api.option('u1').value.get() # assert 'u1' in values._p_.get_cached() # assert 'u1' in settings._p_.get_cached() # c.cfgimpl_reset_cache(True) # assert 'u1' in values._p_.get_cached() # assert 'u1' in settings._p_.get_cached() # sleep(1) # api.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() # c.cfgimpl_reset_cache(True) # 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_not_cache(): od1 = make_description() c = Config(od1) api = getapi(c) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() api.property.pop('cache') api.option('u1').value.get() assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() #def test_reset_cache_only(): # od1 = make_description() # c = Config(od1) # api = getapi(c) # values = c.cfgimpl_get_values() # settings = c.cfgimpl_get_settings() # api.option('u1').value.get() # assert 'u1' in values._p_.get_cached() # assert 'u1' in settings._p_.get_cached() # c.cfgimpl_reset_cache(only=('values',)) # assert 'u1' not in values._p_.get_cached() # assert 'u1' in settings._p_.get_cached() # api.option('u1').value.get() # assert 'u1' in values._p_.get_cached() # assert 'u1' in settings._p_.get_cached() # c.cfgimpl_reset_cache(only=('settings',)) # assert 'u1' in values._p_.get_cached() # assert 'u1' not in settings._p_.get_cached() #def test_force_cache(): # u1 = IntOption('u1', '', multi=True) # u2 = IntOption('u2', '') # u3 = IntOption('u3', '', multi=True) # u4 = IntOption('u4', '', properties=('disabled',)) # od = OptionDescription('od1', '', [u1, u2, u3, u4]) # c = Config(od) # api = getapi(c) # api.property.read_write() # api.property.pop('expire') # api.property.pop('disabled') # # c.cfgimpl_get_values().force_cache() # assert c.cfgimpl_get_values()._p_.get_cached() == {'u1': {None: ([], None)}, # 'u2': {None: (None, None)}, # 'u3': {None: ([], None)}, # 'u4': {None: (None, None)}} # assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'u1': {None: (set(['empty']), None)}, # 'u2': {None: (set([]), None)}, # 'u3': {None: (set(['empty']), None)}, # 'u4': {None: (set(['disabled']), None)}} # api.property.read_only() # # c.cfgimpl_get_values().force_cache() # assert c.cfgimpl_get_values()._p_.get_cached() == {'u1': {None: ([], None)}, # 'u2': {None: (None, None)}, # 'u3': {None: ([], None)}} # assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'empty', 'everything_frozen', 'frozen', 'mandatory', 'validator', 'warnings']), None)}, # 'u1': {None: (set(['empty']), None)}, # 'u2': {None: (set([]), None)}, # 'u3': {None: (set(['empty']), None)}, # 'u4': {None: (set(['disabled']), None)}} # # c.cfgimpl_get_settings().remove('cache') # raises(ConfigError, "c.cfgimpl_get_values().force_cache()") 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) api = getapi(cfg) api.property.read_write() assert cfg.cfgimpl_get_values()._p_.get_cached() == {} #assert cfg.cfgimpl_get_settings()._p_.get_cached() == {} # api.option('ip_admin_eth0.ip_admin_eth0').value.set(['192.168.1.2']) api.option('ip_admin_eth0.ip_admin_eth0').value.get() api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.get() cache = cfg.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.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]) # api.option('ip_admin_eth0.ip_admin_eth0').value.set(['192.168.1.2', '192.168.1.1']) api.option('ip_admin_eth0.ip_admin_eth0').value.get() api.option('ip_admin_eth0.netmask_admin_eth0', 0).value.get() api.option('ip_admin_eth0.netmask_admin_eth0', 1).value.get() cache = cfg.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.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 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) api = getapi(cfg) api.property.read_write() if TIRAMISU_VERSION == 2: api.property.pop('expire') api.option.make_dict() #assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}} assert cfg.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)}} api.option('val1').value.set('new') #assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val3': {None: ('yes', None)}, 'val5': {None: (['yes'], None)}} api.option.make_dict() #assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}} assert cfg.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)}} api.option('val3').value.set('new2') #assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}, # 'val3': {None: (set([]), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1': {None: ('new', None)}, 'val2': {None: ('new', None)}, 'val4': {None: ('new', None)}, 'val5': {None: (['yes'], None)}} api.option.make_dict() #assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}, # 'val3': {None: (set([]), None)}} assert cfg.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)}} api.option('val4').value.set('new3') #assert cfg.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)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1': {None: ('new', None)}, 'val2': {None: ('new', None)}, 'val3': {None: ('new2', None)}, 'val5': {None: (['yes'], None)}} api.option.make_dict() #assert cfg.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)}} assert cfg.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)}} api.option('val5').value.set([undefined, 'new4']) #assert cfg.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)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1': {None: ('new', None)}, 'val2': {None: ('new', None)}, 'val3': {None: ('new2', None)}, 'val4': {None: ('new3', None)}} api.option.make_dict() #assert cfg.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)}} assert cfg.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) api = getapi(cfg) api.property.read_write() api.option.make_dict() global_props = ['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings'] val1_props = [] val1_val1_props = ['empty'] val1_val2_props = [] if TIRAMISU_VERSION == 2: api.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.cfgimpl_get_settings()._p_.get_cached() == {} assert cfg.cfgimpl_get_values()._p_.get_cached() == {} else: assert cfg.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 assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([], None)}} # api.option('val1.val1').value.set([undefined]) assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {} api.option.make_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)} assert cfg.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} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None], None)}, 'val1.val2': {idx_val2: (val_val2, None)}} api.option('val1.val1').value.set([undefined, undefined]) api.option.make_dict() api.option('val1.val2', 1).value.set('oui') assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}} assert cfg.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.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.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None, None], None)}, # 'val1.val2': {None: ([None, 'oui'], None)}} #else: # assert cfg.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) api = getapi(cfg) api.property.read_write() api.option.make_dict() global_props = ['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings'] val1_props = [] val1_val1_props = ['empty'] val1_val2_props = [] if TIRAMISU_VERSION == 2: api.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.cfgimpl_get_settings()._p_.get_cached() == {} assert cfg.cfgimpl_get_values()._p_.get_cached() == {} else: assert cfg.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)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([], None)}} api.option('val1.val1').value.set([undefined]) assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {} api.option.make_dict() #FIXMEassert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'val1': {None: (set([]), None)}} #FIXMEassert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None], None)}, # 'val1.val2': {None: ([None], None)} # } #def test_cache_master_slave_different(): # b = IntOption('int', 'Test int option', default=[0], multi=True) # c = StrOption('str', 'Test string option', multi=True) # d = StrOption('str1', 'Test string option', requires=[{'option': c, 'expected': None, 'action': 'hidden', 'inverse': True}], multi=True) # descr = MasterSlaves("int", "", [b, c, d]) # #descr.impl_set_group_type(groups.master) # cfg = Config(descr) # api = getapi(cfg) # api.property.read_write() # api.property.pop('expire') # api.option.make_dict() # assert cfg.cfgimpl_get_values()._p_.get_cached() == {'int': {None: ([0], None)}, # 'str': {None: ([None], None)}, # 'str1': {None: ([None], None)}} # assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'int': {None: (set(['empty']), None)}, # 'str': {None: (set([]), None), 0: (set([]), None)}, # 'str1': {None: (set([]), None), 0: (set([]), None)}} # api.option('int').value.set([0, 1]) # api.option.make_dict() # assert cfg.cfgimpl_get_values()._p_.get_cached() == {'int': {None: ([0, 1], None)}, # 'str': {None: ([None, None], None)}, # 'str1': {None: ([None, None], None)}} # assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'int': {None: (set(['empty']), None)}, # 'str': {None: (set([]), None), 0: (set([]), None), 1: (set([]), None)}, # 'str1': {None: (set([]), None), 0: (set([]), None), 1: (set([]), None)}} # # api.option('str', 1).value.set('1') # api.option.make_dict() # assert set(cfg.cfgimpl_get_values()._p_.get_cached().keys()) == set(['int', 'str', 'str1']) # assert cfg.cfgimpl_get_values()._p_.get_cached()['int'] == {None: ([0, 1], None)} # assert cfg.cfgimpl_get_values()._p_.get_cached()['str'] == {None: ([None, '1'], None)} # assert cfg.cfgimpl_get_values()._p_.get_cached()['str1'][None][0][0] == None # raises(PropertiesOptionError, "cfg.cfgimpl_get_values()._p_.get_cached()['str1'][None][0][1]") # assert cfg.cfgimpl_get_values()._p_.get_cached()['str1'][None][1] == None # assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, # 'int': {None: (set(['empty']), None)}, # 'str': {None: (set([]), None), 0: (set([]), None), 1: (set([]), None)}, # 'str1': {None: (set([]), None), 0: (set([]), None), 1: (set(['hidden']), None)}} # api.property.read_only() # assert cfg.cfgimpl_get_values()._p_.get_cached() == {} # assert cfg.cfgimpl_get_settings()._p_.get_cached() == {} # api.option.make_dict() # assert cfg.cfgimpl_get_values()._p_.get_cached() == {'int': {None: ([0, 1], None)}, # 'str': {None: ([None, '1'], None)}, # 'str1': {None: ([None, None], None)}} # assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'everything_frozen', 'validator', 'warnings', 'empty', 'mandatory', ]), None)}, # 'int': {None: (set(['empty']), None)}, # 'str': {None: (set([]), None), 0: (set([]), None), 1: (set([]), None)}, # 'str1': {None: (set([]), None), 0: (set([]), None), 1: (set(['hidden']), 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]) c = Config(od) api = getapi(c) api.property.read_write() if TIRAMISU_VERSION == 2: api.property.pop('expire') #assert c.cfgimpl_get_settings()._p_.get_cached() == {} assert c.cfgimpl_get_values()._p_.get_cached() == {} assert api.option('ip_address_service').value.get() == None assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'activate_service': {None: (set([]), None)}, 'ip_address_service': {None: (set([]), None)}} if TIRAMISU_VERSION == 2: assert c.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}} else: assert c.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}, 'activate_service': {None: (True, None)}} api.option.make_dict() assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'activate_service': {None: (set([]), None)}, 'ip_address_service': {None: (set([]), None)}} assert c.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}, 'activate_service': {None: (True, None)}} api.option('ip_address_service').value.set('1.1.1.1') assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'activate_service': {None: (set([]), None)}} assert c.cfgimpl_get_values()._p_.get_cached() == {'activate_service': {None: (True, None)}} api.option.make_dict() assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'activate_service': {None: (set([]), None)}, 'ip_address_service': {None: (set([]), None)}} assert c.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: ('1.1.1.1', None)}, 'activate_service': {None: (True, None)}} api.option('activate_service').value.set(False) assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}} assert c.cfgimpl_get_values()._p_.get_cached() == {} api.option.make_dict() assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'activate_service': {None: (set([]), None)}, 'ip_address_service': {None: (set(['disabled']), None)}} assert c.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]) c = Config(od) api = getapi(c) api.property.read_write() if TIRAMISU_VERSION == 2: api.property.pop('expire') #assert c.cfgimpl_get_settings()._p_.get_cached() == {} assert c.cfgimpl_get_values()._p_.get_cached() == {} assert api.option('ip_address_service').value.get() == None assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'activate_service': {None: (set([]), None)}, 'ip_address_service': {None: (set([]), None)}} if TIRAMISU_VERSION == 2: assert c.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}} else: assert c.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}, 'activate_service': {None: (True, None)}} api.property.pop('disabled') assert api.option('ip_address_service').value.get() == None assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'activate_service': {None: (set([]), None)}, 'ip_address_service': {None: (set([]), None)}} api.property.add('test') assert api.option('ip_address_service').value.get() == None assert c.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'frozen', 'hidden', 'validator', 'warnings', 'test']), 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) api = getapi(cfg) api.property.read_write() assert api.option('val1').value.get() == 1 sleep(1) assert api.option('val2').value.get() == 1 sleep(1) assert api.option('val1').value.get() == 1 assert api.option('val2').value.get() == 1 sleep(2) assert api.option('val1').value.get() == 2 assert api.option('val2').value.get() == 2 assert api.option('val1').value.get() == 2 assert api.option('val2').value.get() == 2