# 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 time import sleep, time from py.test import raises global incr incr = 0 def return_incr(): global incr incr += 1 return incr 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) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() c.u1 assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() c.u2 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) 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 c.u1 == [100] def test_get_cache_no_expire(): # force a value in cache, try if reget corrupted value od1 = make_description() c = Config(od1) 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 c.u1 == [200] def test_cache_reset(): od1 = make_description() c = Config(od1) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() #when change a value c.u1 c.u2 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.u2 = 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' in settings._p_.get_cached() #when remove a value c.u1 assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() del(c.u2) assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u2' not in values._p_.get_cached() assert 'u2' in settings._p_.get_cached() #when add/del property c.u1 assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() c.cfgimpl_get_settings()[od1.u2].append('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() c.u1 assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() c.cfgimpl_get_settings()[od1.u2].remove('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 c.u1 assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() c.cfgimpl_get_settings().append('test') assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() c.u1 assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() c.cfgimpl_get_settings().remove('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) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() c.u1 c.u3 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 c.u3 = [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' in settings._p_.get_cached() #when append value c.u1 c.u3 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() c.u3.append(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' in settings._p_.get_cached() #when pop value c.u1 c.u3 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() c.u3.pop(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' in settings._p_.get_cached() #when remove a value c.u1 assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() del(c.u3) assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() assert 'u3' not in values._p_.get_cached() assert 'u3' in settings._p_.get_cached() def test_reset_cache(): od1 = make_description() c = Config(od1) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() c.u1 assert 'u1' in values._p_.get_cached() assert 'u1' in settings._p_.get_cached() c.cfgimpl_reset_cache() assert 'u1' not in values._p_.get_cached() assert 'u1' not in settings._p_.get_cached() c.u1 sleep(1) c.u1 sleep(1) c.u2 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() 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) values = c.cfgimpl_get_values() c.od1.u1 assert 'od1.u1' in values._p_.get_cached() c.od1.cfgimpl_reset_cache() assert 'od1.u1' not in values._p_.get_cached() def test_reset_cache_only_expired(): od1 = make_description() c = Config(od1) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() c.u1 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) c.u1 sleep(1) c.u2 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) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() settings.remove('expire') c.u1 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) c.u2 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) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() settings.remove('cache') c.u1 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) values = c.cfgimpl_get_values() settings = c.cfgimpl_get_settings() c.u1 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() c.u1 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) c.cfgimpl_get_settings().remove('expire') 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', 'validator', 'warnings']), None)}, 'u1': {None: (set(['empty']), None)}, 'u2': {None: (set([]), None)}, 'u3': {None: (set(['empty']), None)}, 'u4': {None: (set(['disabled']), None)}} c.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) cfg.read_write() assert cfg.cfgimpl_get_values()._p_.get_cached() == {} assert cfg.cfgimpl_get_settings()._p_.get_cached() == {} # cfg.ip_admin_eth0.ip_admin_eth0.append('192.168.1.2') cfg.ip_admin_eth0.ip_admin_eth0 cfg.ip_admin_eth0.netmask_admin_eth0 cache = cfg.cfgimpl_get_values()._p_.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 = 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]) assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([None, 0]) # cfg.ip_admin_eth0.ip_admin_eth0.append('192.168.1.1') cfg.ip_admin_eth0.ip_admin_eth0 cfg.ip_admin_eth0.netmask_admin_eth0 cache = cfg.cfgimpl_get_values()._p_.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 = 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]) assert set(cache['ip_admin_eth0.netmask_admin_eth0'].keys()) == set([None, 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={'': ((val1, False),)}, properties=('mandatory',)) val3 = StrOption('val3', "", callback=return_value, callback_params={'': ('yes',)}) val4 = StrOption('val4', "", callback=return_value, callback_params={'value': ((val1, False),)}) val5 = StrOption('val5', "", callback=return_value, callback_params={'value': ('yes',)}, multi=True) maconfig = OptionDescription('rootconfig', '', [val1, val2, val3, val4, val5]) cfg = Config(maconfig) cfg.cfgimpl_get_settings().remove('expire') cfg.read_write() cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val2': {None: (set(['mandatory']), None)}, 'val3': {None: (set([]), None)}, 'val4': {None: (set([]), None)}, 'val5': {None: (set(['empty']), 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)}} cfg.val1 = 'new' 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)}, 'val5': {None: (set(['empty']), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val3': {None: ('yes', None)}, 'val5': {None: (['yes'], None)}} cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val2': {None: (set(['mandatory']), 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: ('yes', None)}, 'val4': {None: ('new', None)}, 'val5': {None: (['yes'], None)}} cfg.val3 = 'new2' assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val2': {None: (set(['mandatory']), 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)}, 'val4': {None: ('new', None)}, 'val5': {None: (['yes'], None)}} cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val2': {None: (set(['mandatory']), 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: ('new', None)}, 'val5': {None: (['yes'], None)}} cfg.val4 = 'new3' assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val2': {None: (set(['mandatory']), 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)}, 'val5': {None: (['yes'], None)}} cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val2': {None: (set(['mandatory']), 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'], None)}} cfg.val5.append('new4') assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val2': {None: (set(['mandatory']), 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)}} cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val2': {None: (set(['mandatory']), 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) cfg.cfgimpl_get_settings().remove('expire') cfg.read_write() cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val1.val1': {None: (set(['empty']), None)}, 'val1.val2': {None: (set([]), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([], None)}, 'val1.val2': {None: ([], None)}} cfg.val1.val1.append() 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() == {} cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val1.val1': {None: (set(['empty']), None)}, 'val1.val2': {None: (set([]), None), 0: (set([]), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None], None)}, 'val1.val2': {None: ([None], None)}} cfg.val1.val1.append() cfg.cfgimpl_get_values().force_cache() cfg.val1.val2[1] = '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() == {} cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val1.val1': {None: (set(['empty']), None)}, 'val1.val2': {None: (set([]), None), 0: (set([]), None), 1: (set([]), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None, None], None)}, 'val1.val2': {None: ([None, 'oui'], None)}} def test_cache_master_callback(): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", multi=True, callback=return_value, callback_params={'value': ((val1, False),)}) interface1 = MasterSlaves('val1', '', [val1, val2]) #interface1.impl_set_group_type(groups.master) maconfig = OptionDescription('rootconfig', '', [interface1]) cfg = Config(maconfig) cfg.cfgimpl_get_settings().remove('expire') cfg.read_write() cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val1.val1': {None: (set(['empty']), None)}, 'val1.val2': {None: (set([]), None)}} assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([], None)}, 'val1.val2': {None: ([], None)}} cfg.val1.val1.append() 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() == {} cfg.cfgimpl_get_values().force_cache() assert cfg.cfgimpl_get_settings()._p_.get_cached() == {None: {None: (set(['cache', 'disabled', 'frozen', 'hidden', 'validator', 'warnings']), None)}, 'val1': {None: (set([]), None)}, 'val1.val1': {None: (set(['empty']), None)}, 'val1.val2': {None: (set([]), None), 0: (set([]), None)}} #assert cfg.cfgimpl_get_values()._p_.get_cached() == {'val1.val1': {None: ([None], None)}, # 'val1.val2': {None: ([None], None), 0: (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) cfg.cfgimpl_get_settings().remove('expire') cfg.read_write() cfg.cfgimpl_get_values().force_cache() 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)}} cfg.int = [0, 1] cfg.cfgimpl_get_values().force_cache() 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)}} cfg.str = [None, '1'] cfg.cfgimpl_get_values().force_cache() 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)}} cfg.read_only() assert cfg.cfgimpl_get_values()._p_.get_cached() == {} assert cfg.cfgimpl_get_settings()._p_.get_cached() == {} cfg.cfgimpl_get_values().force_cache() 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) c.cfgimpl_get_settings().remove('expire') c.read_write() assert c.cfgimpl_get_settings()._p_.get_cached() == {} assert c.cfgimpl_get_values()._p_.get_cached() == {} assert c.ip_address_service == 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)}} assert c.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}} c.cfgimpl_get_values().force_cache() 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)}} c.ip_address_service = '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)}, 'ip_address_service': {None: (set([]), None)}} assert c.cfgimpl_get_values()._p_.get_cached() == {'activate_service': {None: (True, None)}} c.cfgimpl_get_values().force_cache() 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)}} c.activate_service = False 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() == {} c.cfgimpl_get_values().force_cache() 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) c.cfgimpl_get_settings().remove('expire') c.read_write() assert c.cfgimpl_get_settings()._p_.get_cached() == {} assert c.cfgimpl_get_values()._p_.get_cached() == {} assert c.ip_address_service == 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)}} assert c.cfgimpl_get_values()._p_.get_cached() == {'ip_address_service': {None: (None, None)}} c.cfgimpl_get_settings().remove('disabled') assert c.ip_address_service == 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)}} c.cfgimpl_get_settings().append('test') assert c.ip_address_service == 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) val2 = IntOption('val2', "", callback=return_value, callback_params={'value': ((val1, False),)}) maconfig = OptionDescription('rootconfig', '', [val1, val2]) cfg = Config(maconfig) cfg.read_write() assert cfg.val1 == 1 sleep(1) assert cfg.val2 == 1 sleep(1) assert cfg.val1 == 2 assert cfg.val2 == 2