from tiramisu.option import BoolOption, UnicodeOption, SymLinkOption, \ OptionDescription from tiramisu.config import Config from tiramisu.setting import owners from tiramisu.storage import delete_session from tiramisu.error import ConfigError from pickle import dumps, loads def return_value(value=None): return value def _get_slots(opt): slots = set() for subclass in opt.__class__.__mro__: if subclass is not object: slots.update(subclass.__slots__) return slots def _no_state(opt): for attr in _get_slots(opt): if 'state' in attr: try: getattr(opt, attr) except: pass else: raise Exception('opt should have already attribute {0}'.format(attr)) def _diff_opt(opt1, opt2): attr1 = set(_get_slots(opt1)) attr2 = set(_get_slots(opt2)) diff1 = attr1 - attr2 diff2 = attr2 - attr1 if diff1 != set(): raise Exception('more attribute in opt1 {0}'.format(list(diff1))) if diff2 != set(): raise Exception('more attribute in opt2 {0}'.format(list(diff2))) for attr in attr1: if attr in ['_cache_paths', '_cache_consistencies']: continue err1 = False err2 = False val1 = None val2 = None try: val1 = getattr(opt1, attr) except: err1 = True try: val2 = getattr(opt2, attr) except: err2 = True assert err1 == err2 if val1 is None: assert val1 == val2 elif attr == '_children': assert val1[0] == val2[0] for index, _opt in enumerate(val1[1]): assert _opt._name == val2[1][index]._name elif attr == '_requires': assert val1[0][0][0]._name == val2[0][0][0]._name assert val1[0][0][1:] == val2[0][0][1:] elif attr == '_opt': assert val1._name == val2._name elif attr == '_consistencies': # dict is only a cache if isinstance(val1, list): for index, consistency in enumerate(val1): assert consistency[0] == val2[index][0] for idx, opt in enumerate(consistency[1]): assert opt._name == val2[index][1][idx]._name elif attr == '_callback': assert val1[0] == val2[0] if val1[1] is not None: for key, values in val1[1].items(): for idx, value in enumerate(values): if isinstance(value, tuple): assert val1[1][key][idx][0]._name == val2[1][key][idx][0]._name assert val1[1][key][idx][1] == val2[1][key][idx][1] else: assert val1[1][key][idx] == val2[1][key][idx] else: assert val1[1] == val2[1] else: assert val1 == val2 def test_diff_opt(): b = BoolOption('b', '') u = UnicodeOption('u', '', requires=[{'option': b, 'expected': True, 'action': 'disabled', 'inverse': True}]) #u.impl_add_consistency('not_equal', b) s = SymLinkOption('s', u) o = OptionDescription('o', '', [b, u, s]) o1 = OptionDescription('o1', '', [o]) a = dumps(o1) q = loads(a) _diff_opt(o1, q) _diff_opt(o1.o, q.o) _diff_opt(o1.o.b, q.o.b) _diff_opt(o1.o.u, q.o.u) _diff_opt(o1.o.s, q.o.s) def test_diff_opt_cache(): b = BoolOption('b', '') u = UnicodeOption('u', '', requires=[{'option': b, 'expected': True, 'action': 'disabled', 'inverse': True}]) u.impl_add_consistency('not_equal', b) s = SymLinkOption('s', u) o = OptionDescription('o', '', [b, u, s]) o1 = OptionDescription('o1', '', [o]) o1.impl_build_cache() a = dumps(o1) q = loads(a) _diff_opt(o1, q) _diff_opt(o1.o, q.o) _diff_opt(o1.o.b, q.o.b) _diff_opt(o1.o.u, q.o.u) _diff_opt(o1.o.s, q.o.s) def test_diff_opt_callback(): b = BoolOption('b', '', callback=return_value) b2 = BoolOption('b2', '', callback=return_value, callback_params={'': ('yes',)}) b3 = BoolOption('b3', '', callback=return_value, callback_params={'': ('yes', (b, False)), 'value': ('no',)}) o = OptionDescription('o', '', [b, b2, b3]) o1 = OptionDescription('o1', '', [o]) o1.impl_build_cache() a = dumps(o1) q = loads(a) _diff_opt(o1, q) _diff_opt(o1.o, q.o) _diff_opt(o1.o.b, q.o.b) _diff_opt(o1.o.b2, q.o.b2) _diff_opt(o1.o.b3, q.o.b3) def test_no_state_attr(): # all _state_xxx attributes should be deleted b = BoolOption('b', '') u = UnicodeOption('u', '', requires=[{'option': b, 'expected': True, 'action': 'disabled', 'inverse': True}]) s = SymLinkOption('s', u) o = OptionDescription('o', '', [b, u, s]) o1 = OptionDescription('o1', '', [o]) a = dumps(o1) q = loads(a) _no_state(q) _no_state(q.o) _no_state(q.o.b) _no_state(q.o.u) _no_state(q.o.s) def test_state_config(): val1 = BoolOption('val1', "") maconfig = OptionDescription('rootconfig', '', [val1]) try: cfg = Config(maconfig, persistent=True, session_id='29090931') except ValueError: cfg = Config(maconfig, session_id='29090931') cfg._impl_test = True a = dumps(cfg) q = loads(a) _diff_opt(maconfig, q.cfgimpl_get_description()) assert cfg.cfgimpl_get_values().get_modified_values() == q.cfgimpl_get_values().get_modified_values() assert cfg.cfgimpl_get_settings().get_modified_properties() == q.cfgimpl_get_settings().get_modified_properties() assert cfg.cfgimpl_get_settings().get_modified_permissives() == q.cfgimpl_get_settings().get_modified_permissives() try: delete_session('29090931') except ConfigError: pass def test_state_properties(): val1 = BoolOption('val1', "") maconfig = OptionDescription('rootconfig', '', [val1]) try: cfg = Config(maconfig, persistent=True, session_id='29090932') except ValueError: cfg = Config(maconfig, session_id='29090932') cfg._impl_test = True cfg.read_write() cfg.cfgimpl_get_settings()[val1].append('test') a = dumps(cfg) q = loads(a) _diff_opt(maconfig, q.cfgimpl_get_description()) assert cfg.cfgimpl_get_values().get_modified_values() == q.cfgimpl_get_values().get_modified_values() assert cfg.cfgimpl_get_settings().get_modified_properties() == q.cfgimpl_get_settings().get_modified_properties() assert cfg.cfgimpl_get_settings().get_modified_permissives() == q.cfgimpl_get_settings().get_modified_permissives() try: delete_session('29090931') except ConfigError: pass def test_state_values(): val1 = BoolOption('val1', "") maconfig = OptionDescription('rootconfig', '', [val1]) try: cfg = Config(maconfig, persistent=True, session_id='29090933') except ValueError: cfg = Config(maconfig, session_id='29090933') cfg._impl_test = True cfg.val1 = True a = dumps(cfg) q = loads(a) _diff_opt(maconfig, q.cfgimpl_get_description()) assert cfg.cfgimpl_get_values().get_modified_values() == q.cfgimpl_get_values().get_modified_values() assert cfg.cfgimpl_get_settings().get_modified_properties() == q.cfgimpl_get_settings().get_modified_properties() assert cfg.cfgimpl_get_settings().get_modified_permissives() == q.cfgimpl_get_settings().get_modified_permissives() q.val1 = False #assert cfg.val1 is True assert q.val1 is False try: delete_session('29090931') except ConfigError: pass def test_state_values_owner(): val1 = BoolOption('val1', "") maconfig = OptionDescription('rootconfig', '', [val1]) try: cfg = Config(maconfig, persistent=True, session_id='29090934') except ValueError: cfg = Config(maconfig, session_id='29090934') cfg._impl_test = True owners.addowner('newowner') cfg.cfgimpl_get_settings().setowner(owners.newowner) cfg.val1 = True a = dumps(cfg) q = loads(a) _diff_opt(maconfig, q.cfgimpl_get_description()) assert cfg.cfgimpl_get_values().get_modified_values() == q.cfgimpl_get_values().get_modified_values() assert cfg.cfgimpl_get_settings().get_modified_properties() == q.cfgimpl_get_settings().get_modified_properties() assert cfg.cfgimpl_get_settings().get_modified_permissives() == q.cfgimpl_get_settings().get_modified_permissives() q.val1 = False nval1 = q.cfgimpl_get_description().val1 assert q.getowner(nval1) == owners.newowner try: delete_session('29090931') except ConfigError: pass def test_state_unkown_setting_owner(): """load an unknow _owner, should create it""" assert not 'supernewuser' in owners.__dict__ loads("""ccopy_reg _reconstructor p0 (ctiramisu.setting Settings p1 c__builtin__ object p2 Ntp3 Rp4 (dp5 S'_owner' p6 S'supernewuser' p7 sS'_p_' p8 g0 (ctiramisu.storage.dictionary.setting Settings p9 g2 Ntp10 Rp11 (dp12 S'_cache' p13 (dp14 sS'_permissives' p15 (dp16 sS'_properties' p17 (dp18 sbsb. .""") assert 'supernewuser' in owners.__dict__