"""test API """ import weakref import pytest import warnings from copy import copy from py.test import raises from collections import OrderedDict from .autopath import do_autopath do_autopath() from tiramisu import Config, MetaConfig, \ StrOption, SymLinkOption, OptionDescription, Leadership, DynOptionDescription, \ submulti, undefined, owners, Params, ParamOption from tiramisu.error import PropertiesOptionError, APIError, ConfigError, LeadershipError ICON = u'\u2937' OPTIONS_TYPE = {'str': {'type': str, 'option': StrOption} } PROPERTIES = ['hidden', 'disabled', 'mandatory'] PROPERTIES_LIST = ['prop1', 'prop2'] OWNER = 'user' # multi is False FIRST_VALUE = 'myvalue' SECOND_VALUE = 'myvalue1' EMPTY_VALUE = None # multi is True LIST_FIRST_VALUE = ['myvalue'] LIST_SECOND_VALUE = ['myvalue', 'myvalue1'] LIST_EMPTY_VALUE = [] # multi is submulti SUBLIST_FIRST_VALUE = [['myvalue']] SUBLIST_SECOND_VALUE = [['myvalue'], ['myvalue1', 'myvalue2']] SUBLIST_EMPTY_VALUE = [] DISPLAY = True DISPLAY = False def return_list(val=None, suffix=None): if val: return val else: return ['val1', 'val2'] def return_str(val, suffix=None): return val def display_info(func): def wrapper(*args, **kwargs): if DISPLAY: print(u'\n{} {}'.format(ICON, func.__name__)) return func(*args, **kwargs) return wrapper autocheck_registers = [] def autocheck(func): autocheck_registers.append(func) def wrapper(*args, **kwargs): if DISPLAY and kwargs.get('display', True): print(u' {} {}'.format(ICON, func.__name__)) return func(*args, **kwargs) return wrapper def _autocheck_default_value(cfg, path, conf, **kwargs): """set and get values """ # check if is a multi, a leader or a follower multi = cfg.unrestraint.option(path).option.ismulti() submulti_ = cfg.unrestraint.option(path).option.issubmulti() isfollower = cfg.unrestraint.option(path).option.isfollower() # set default value (different if value is multi or not) empty_value = kwargs['default'] # test default value (should be empty) # cannot test for follower (we cannot get all values for a follower) if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg with warnings.catch_warnings(record=True) as w: if not isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(path).value.get() == empty_value assert cfg_.forcepermissive.option(path).value.get() == empty_value elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(path).value.get()") assert cfg_.forcepermissive.option(path).value.get() == empty_value else: raises(PropertiesOptionError, "cfg_.option(path).value.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(path).value.get()") else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(path, 0).value.get() == empty_value assert cfg_.option(path, 1).value.get() == empty_value assert cfg_.forcepermissive.option(path, 0).value.get() == empty_value assert cfg_.forcepermissive.option(path, 1).value.get() == empty_value elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(path, 0).value.get()") assert cfg_.forcepermissive.option(path, 0).value.get() == empty_value assert cfg_.forcepermissive.option(path, 1).value.get() == empty_value else: raises(PropertiesOptionError, "cfg_.option(path, 0).value.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(path, 0).value.get()") def _set_value(cfg, pathwrite, conf, **kwargs): set_permissive = kwargs.get('set_permissive', True) multi = cfg.unrestraint.option(pathwrite).option.ismulti() submulti_ = cfg.unrestraint.option(pathwrite).option.issubmulti() isleader = cfg.unrestraint.option(pathwrite).option.isleader() isfollower = cfg.unrestraint.option(pathwrite).option.isfollower() if not multi: first_value = FIRST_VALUE elif submulti_ is False: if not isfollower: first_value = LIST_FIRST_VALUE else: second_value = LIST_SECOND_VALUE[1] else: if not isfollower: first_value = SUBLIST_FIRST_VALUE else: second_value = SUBLIST_SECOND_VALUE[1] # for follower should have an index and good length # for leader must append, not set if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg with warnings.catch_warnings(record=True) as w: if isleader: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): raises(APIError, "cfg_.option(pathwrite, 0).value.set(first_value[0])") if not set_permissive: cfg_.option(pathwrite).value.set([first_value[0]]) else: cfg_.forcepermissive.option(pathwrite).value.set([first_value[0]]) elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(pathwrite).value.set([first_value[0]])") if set_permissive: cfg_.forcepermissive.option(pathwrite).value.set([first_value[0]]) else: raises(PropertiesOptionError, "cfg_.option(pathwrite).value.set([first_value[0]])") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathwrite).value.set([first_value[0]])") if len(first_value) > 1: raises(APIError, "cfg_.unrestraint.option(pathwrite).value.set(first_value[1])") elif isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): if not set_permissive: cfg_.option(pathwrite, 1).value.set(second_value) else: cfg_.forcepermissive.option(pathwrite, 1).value.set(second_value) elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(pathwrite, 1).value.set(second_value)") if set_permissive: cfg_.forcepermissive.option(pathwrite, 1).value.set(second_value) else: raises(PropertiesOptionError, "cfg_.option(pathwrite, 1).value.set(second_value)") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathwrite, 1).value.set(second_value)") raises(APIError, "cfg_.unrestraint.option(pathwrite).value.set([second_value, second_value])") else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): if not set_permissive: cfg_.option(pathwrite).value.set(first_value) else: cfg_.forcepermissive.option(pathwrite).value.set(first_value) elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(pathwrite).value.set(first_value)") if set_permissive: cfg_.forcepermissive.option(pathwrite).value.set(first_value) else: raises(PropertiesOptionError, "cfg_.option(pathwrite).value.set(first_value)") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathwrite).value.set(first_value)") #FIXME raises(APIError, "cfg_.unrestraint.option(pathwrite).value.set(first_value)") def _getproperties(multi, isfollower, kwargs): # define properties properties = copy(PROPERTIES_LIST) if multi and not isfollower: default_props = ['empty'] properties.append('empty') else: default_props = [] extra_properties = kwargs.get('extra_properties') if extra_properties: properties.extend(extra_properties) default_props.extend(extra_properties) return default_props, frozenset(properties) def _check_properties(cfg, mcfg, pathread, conf, kwargs, props_permissive, props): if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg if not cfg.unrestraint.option(pathread).option.isfollower(): if not kwargs.get('permissive_od', False): assert set(cfg_.option(pathread).property.get()) == set(props_permissive) assert set(cfg_.option(pathread).property.get()) == set(props) else: raises(PropertiesOptionError, "cfg_.option(pathread).property.get()") raises(PropertiesOptionError, "cfg_.option(pathread).property.get()") assert set(cfg_.forcepermissive.option(pathread).property.get()) == set(props_permissive) assert set(cfg_.forcepermissive.option(pathread).property.get()) == set(props) assert set(cfg_.unrestraint.option(pathread).property.get()) == set(props_permissive) assert set(cfg_.unrestraint.option(pathread).property.get()) == set(props) else: if not kwargs.get('permissive_od', False): assert set(cfg_.option(pathread, 0).property.get()) == set(props_permissive) assert set(cfg_.option(pathread, 0).property.get()) == set(props) # assert set(cfg_.option(pathread, 1).property.get()) == set(props_permissive) assert set(cfg_.option(pathread, 1).property.get()) == set(props) else: raises(PropertiesOptionError, "cfg_.option(pathread, 0).property.get()") raises(PropertiesOptionError, "cfg_.option(pathread, 0).property.get()") # raises(PropertiesOptionError, "cfg_.option(pathread, 1).property.get()") raises(PropertiesOptionError, "cfg_.option(pathread, 1).property.get()") assert set(cfg_.forcepermissive.option(pathread, 0).property.get()) == set(props_permissive) assert set(cfg_.forcepermissive.option(pathread, 0).property.get()) == set(props) # assert set(cfg_.forcepermissive.option(pathread, 1).property.get()) == set(props_permissive) assert set(cfg_.forcepermissive.option(pathread, 1).property.get()) == set(props) assert set(cfg_.unrestraint.option(pathread, 1).property.get()) == set(props_permissive) assert set(cfg_.unrestraint.option(pathread, 1).property.get()) == set(props) def _property_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): # check if is a multi or a follower multi = cfg.unrestraint.option(pathread).option.ismulti() isfollower = cfg.unrestraint.option(pathread).option.isfollower() # define properties properties = copy(PROPERTIES_LIST) if multi and not isfollower: default_props = ['empty'] properties.append('empty') else: default_props = [] extra_properties = kwargs.get('extra_properties') if extra_properties: properties.extend(extra_properties) default_props.extend(extra_properties) default_props, properties = _getproperties(multi, isfollower, kwargs) if confwrite == confread: _check_properties(cfg, mcfg, pathread, confwrite, kwargs, default_props, default_props) else: _check_properties(cfg, mcfg, pathread, confread, kwargs, default_props, default_props) # set properties with permissive for prop in properties: if confread is not None: cfg_ = cfg.config(confread) else: cfg_ = cfg if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): cfg_.option(pathwrite).property.add(prop) elif not kwargs.get('propertyerror', False): cfg_.forcepermissive.option(pathwrite).property.add(prop) else: cfg_.unrestraint.option(pathwrite).property.add(prop) if confwrite == confread: _check_properties(cfg, mcfg, pathread, confwrite, kwargs, properties, properties) else: _check_properties(cfg, mcfg, pathread, confread, kwargs, properties, properties) def _autocheck_get_value(cfg, pathread, conf, **kwargs): set_permissive = kwargs.get('set_permissive', True) multi = cfg.unrestraint.option(pathread).option.ismulti() submulti_ = cfg.unrestraint.option(pathread).option.issubmulti() isfollower = cfg.unrestraint.option(pathread).option.isfollower() empty_value = kwargs['default'] if not multi: first_value = FIRST_VALUE elif submulti_ is False: if not isfollower: first_value = LIST_FIRST_VALUE else: second_value = LIST_SECOND_VALUE[1] else: if not isfollower: first_value = SUBLIST_FIRST_VALUE else: second_value = SUBLIST_SECOND_VALUE[1] # get value after set value without permissive if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg with warnings.catch_warnings(record=True) as w: if isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread, 0).value.get() == empty_value assert cfg_.option(pathread, 1).value.get() == second_value assert cfg_.forcepermissive.option(pathread, 0).value.get() == empty_value assert cfg_.forcepermissive.option(pathread, 1).value.get() == second_value elif kwargs.get('permissive', False): raises(PropertiesOptionError, "cfg_.option(pathread, 0).value.get()") assert cfg_.forcepermissive.option(pathread, 0).value.get() == empty_value if set_permissive: assert cfg_.forcepermissive.option(pathread, 1).value.get() == second_value else: assert cfg_.forcepermissive.option(pathread, 1).value.get() == empty_value else: raises(PropertiesOptionError, "cfg_.option(pathread, 0).value.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread, 0).value.get()") else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread).value.get() == first_value assert cfg_.forcepermissive.option(pathread).value.get() == first_value elif kwargs.get('permissive', False): raises(PropertiesOptionError, "cfg_.option(pathread).value.get()") if set_permissive: assert cfg_.forcepermissive.option(pathread).value.get() == first_value else: assert cfg_.forcepermissive.option(pathread).value.get() == empty_value else: raises(PropertiesOptionError, "cfg_.option(pathread).value.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread).value.get()") def _check_owner(cfg, pathread, conf, kwargs, owner, permissive_owner): isfollower = cfg.unrestraint.option(pathread).option.isfollower() if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg if not isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread).owner.get() == owner assert cfg_.forcepermissive.option(pathread).owner.get() == owner elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(pathread).owner.get()") assert cfg_.forcepermissive.option(pathread).owner.get() == permissive_owner else: raises(PropertiesOptionError, "cfg_.option(pathread).owner.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread).owner.get()") else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread, 0).owner.get() == 'default' assert cfg_.forcepermissive.option(pathread, 0).owner.get() == 'default' assert cfg_.option(pathread, 1).owner.get() == owner assert cfg_.forcepermissive.option(pathread, 1).owner.get() == owner elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(pathread, 0).owner.get()") raises(PropertiesOptionError, "cfg_.option(pathread, 1).owner.get()") assert cfg_.forcepermissive.option(pathread, 0).owner.get() == 'default' assert cfg_.forcepermissive.option(pathread, 1).owner.get() == permissive_owner else: raises(PropertiesOptionError, "cfg_.option(pathread, 0).owner.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread, 0).owner.get()") @autocheck def autocheck_option_multi(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): if not kwargs.get('permissive_od', False): cfg.option(pathread).option.ismulti() cfg.option(pathread).option.issubmulti() cfg.option(pathread).option.isleader() cfg.option(pathread).option.isfollower() else: raises(PropertiesOptionError, "cfg.option(pathread).option.ismulti()") raises(PropertiesOptionError, "cfg.option(pathread).option.issubmulti()") raises(PropertiesOptionError, "cfg.option(pathread).option.isleader()") raises(PropertiesOptionError, "cfg.option(pathread).option.isfollower()") cfg.forcepermissive.option(pathread).option.ismulti() cfg.forcepermissive.option(pathread).option.issubmulti() cfg.forcepermissive.option(pathread).option.isleader() cfg.forcepermissive.option(pathread).option.isfollower() @autocheck def autocheck_default_owner(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): """check different value of owner when any value is set to this option """ isfollower = cfg.unrestraint.option(pathread).option.isfollower() # check if owner is a string "default" and 'isdefault' def do(conf): if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg if not isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread).owner.get() == 'default' assert cfg_.forcepermissive.option(pathread).owner.get() == 'default' # assert cfg_.option(pathread).owner.isdefault() assert cfg_.forcepermissive.option(pathread).owner.isdefault() elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(pathread).owner.get()") assert cfg_.forcepermissive.option(pathread).owner.get() == 'default' # raises(PropertiesOptionError, "cfg_.option(pathread).owner.isdefault()") assert cfg_.forcepermissive.option(pathread).owner.isdefault() else: raises(PropertiesOptionError, "cfg_.option(pathread).owner.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread).owner.get()") # raises(PropertiesOptionError, "cfg_.option(pathread).owner.isdefault()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread).owner.isdefault()") # assert cfg_.unrestraint.option(pathread).owner.get() == 'default' assert cfg_.unrestraint.option(pathread).owner.isdefault() else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread, 0).owner.get() == 'default' assert cfg_.forcepermissive.option(pathread, 0).owner.get() == 'default' # assert cfg_.option(pathread, 0).owner.isdefault() assert cfg_.forcepermissive.option(pathread, 0).owner.isdefault() elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg_.option(pathread, 0).owner.get()") assert cfg_.forcepermissive.option(pathread, 0).owner.get() == 'default' # raises(PropertiesOptionError, "cfg_.option(pathread, 0).owner.isdefault()") assert cfg_.forcepermissive.option(pathread, 0).owner.isdefault() else: raises(PropertiesOptionError, "cfg_.option(pathread, 0).owner.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread, 0).owner.get()") # raises(PropertiesOptionError, "cfg_.option(pathread, 0).owner.isdefault()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread, 0).owner.isdefault()") assert cfg_.unrestraint.option(pathread, 0).owner.get() == 'default' assert cfg_.unrestraint.option(pathread, 0).owner.isdefault() do(confread) if confread != confwrite: do(confwrite) @autocheck def autocheck_default_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): _autocheck_default_value(cfg, pathread, confread, **kwargs) if confread != confwrite: _autocheck_default_value(cfg, pathread, confwrite, **kwargs) @autocheck def autocheck_set_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): _set_value(cfg, pathwrite, confwrite, **kwargs) @autocheck def autocheck_get_value_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): multi = cfg.unrestraint.option(pathread).option.ismulti() submulti_ = cfg.unrestraint.option(pathread).option.issubmulti() isfollower = cfg.unrestraint.option(pathread).option.isfollower() _set_value(cfg, pathwrite, confwrite, **kwargs) empty_value = kwargs['default'] if not multi: first_value = FIRST_VALUE elif submulti_ is False: first_value = LIST_FIRST_VALUE else: first_value = SUBLIST_FIRST_VALUE def do(conf): # get value after set value without permissive if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg if isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread, 0).value.get() == empty_value assert cfg_.forcepermissive.option(pathread, 0).value.get() == empty_value if submulti_: assert cfg_.option(pathread, 1).value.get() == SUBLIST_SECOND_VALUE[1] assert cfg_.forcepermissive.option(pathread, 1).value.get() == SUBLIST_SECOND_VALUE[1] else: assert cfg_.option(pathread, 1).value.get() == LIST_SECOND_VALUE[1] assert cfg_.forcepermissive.option(pathread, 1).value.get() == LIST_SECOND_VALUE[1] elif kwargs.get('permissive', False): raises(PropertiesOptionError, "assert cfg_.option(pathread, 0).value.get()") raises(PropertiesOptionError, "assert cfg_.option(pathread, 1).value.get()") assert cfg_.forcepermissive.option(pathread, 0).value.get() == empty_value if submulti_: assert cfg_.forcepermissive.option(pathread, 1).value.get() == SUBLIST_SECOND_VALUE[1] else: assert cfg_.forcepermissive.option(pathread, 1).value.get() == LIST_SECOND_VALUE[1] else: raises(PropertiesOptionError, "assert cfg_.option(pathread, 0).value.get()") raises(PropertiesOptionError, "assert cfg_.option(pathread, 1).value.get()") raises(PropertiesOptionError, "assert cfg_.forcepermissive.option(pathread, 0).value.get()") raises(PropertiesOptionError, "assert cfg_.forcepermissive.option(pathread, 1).value.get()") else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): with warnings.catch_warnings(record=True) as w: assert cfg_.option(pathread).value.get() == first_value assert cfg_.forcepermissive.option(pathread).value.get() == first_value elif kwargs.get('permissive', False): raises(PropertiesOptionError, "cfg_.option(pathread).value.get()") assert cfg_.forcepermissive.option(pathread).value.get() == first_value else: raises(PropertiesOptionError, "cfg_.option(pathread).value.get()") raises(PropertiesOptionError, "cfg_.forcepermissive.option(pathread).value.get()") with warnings.catch_warnings(record=True) as w: do(confread) if confread != confwrite: do(confwrite) @autocheck def autocheck_get_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): _set_value(cfg, pathwrite, confwrite, set_permissive=False, **kwargs) _autocheck_get_value(cfg, pathread, confread, set_permissive=False, **kwargs) if pathread.endswith('val1'): val2_path = pathread.replace('val1', 'val2') _autocheck_default_value(cfg, val2_path, confread, **kwargs) if confread != confwrite: _autocheck_get_value(cfg, pathread, confwrite, set_permissive=False, **kwargs) if pathread.endswith('val1'): _autocheck_default_value(cfg, val2_path, confwrite, **kwargs) @autocheck def autocheck_value_follower(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): isfollower = cfg.unrestraint.option(pathread).option.isfollower() if not isfollower: return if kwargs.get('propertyerror', False): return submulti_ = cfg.forcepermissive.option(pathread).option.issubmulti() empty_value = kwargs['default'] def do(conf): if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg length = cfg_.option(pathread).value.len() assert cfg_.forcepermissive.option(pathread).value.len() == length assert length == 2 do(confread) if confread != confwrite: do(confwrite) length = 2 value = [] with warnings.catch_warnings(record=True) as w: for idx in range(length): value.append(cfg.forcepermissive.option(pathread, idx).value.get()) assert value == [empty_value, empty_value] # cannot access to a follower with index too high if submulti_ is False: value = LIST_FIRST_VALUE[0] else: value = SUBLIST_FIRST_VALUE[0] raises(LeadershipError, "cfg.forcepermissive.option(pathread, length).value.get()") raises(LeadershipError, "cfg.forcepermissive.option(pathread, length).value.set(value)") raises(LeadershipError, "cfg.forcepermissive.option(pathread, length).value.reset()") raises(LeadershipError, "cfg.forcepermissive.option(pathread, length).owner.get()") raises(LeadershipError, "cfg.forcepermissive.option(pathread, length).owner.isdefault()") raises(LeadershipError, "cfg.forcepermissive.option(pathread, length).property.get()") raises(LeadershipError, "cfg.forcepermissive.option(pathread, length).owner.set('new_user')") raises(LeadershipError, "cfg.forcepermissive.option(pathread, length).property.add('prop')") @autocheck def autocheck_reset_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): # check if is a multi, a leader or a follower multi = cfg.unrestraint.option(pathread).option.ismulti() submulti_ = cfg.unrestraint.option(pathread).option.issubmulti() isfollower = cfg.unrestraint.option(pathread).option.isfollower() # set default value (different if value is multi or not) if not multi: first_value = FIRST_VALUE second_value = SECOND_VALUE elif submulti_ is False: first_value = LIST_FIRST_VALUE second_value = LIST_SECOND_VALUE else: first_value = SUBLIST_FIRST_VALUE second_value = SUBLIST_SECOND_VALUE empty_value = kwargs['default'] _set_value(cfg, pathwrite, confwrite, **kwargs) # reset value without permissive with warnings.catch_warnings(record=True) as w: if confwrite is not None: cfg_ = cfg.config(confwrite) else: cfg_ = cfg if not isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): cfg_.option(pathwrite).value.reset() #else: #FIXME raises(PropertiesOptionError, "cfg.config(confwrite).option(pathwrite).value.reset()") else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): cfg_.option(pathwrite, 0).value.reset() #else: #FIXME raises(PropertiesOptionError, "cfg.config(confwrite).option(pathwrite, 0).value.reset()") # get value after reset value without permissive def do(conf): if confwrite is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg if isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread, 0).value.get() == empty_value assert cfg_.option(pathread, 1).value.get() == second_value[1] elif kwargs.get('permissive', False): raises(PropertiesOptionError, "cfg_.option(pathread, 0).value.get()") assert cfg_.forcepermissive.option(pathread, 0).value.get() == empty_value assert cfg_.forcepermissive.option(pathread, 1).value.get() == second_value[1] else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread).value.get() == empty_value elif kwargs.get('permissive', False): raises(PropertiesOptionError, "cfg_.option(pathread).value.get()") assert cfg_.forcepermissive.option(pathread).value.get() == first_value with warnings.catch_warnings(record=True) as w: do(confread) if confread != confwrite: do(confwrite) @autocheck def autocheck_append_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): isleader = cfg.unrestraint.option(pathread).option.isleader() submulti_ = cfg.unrestraint.option(pathread).option.issubmulti() if not isleader: return if confread is not None: cfg_ = cfg.config(confread) else: cfg_ = cfg if confwrite is not None: cfg2_ = cfg.config(confwrite) else: cfg2_ = cfg with warnings.catch_warnings(record=True) as w: if not kwargs.get('propertyerror', False): leader_value = cfg_.forcepermissive.option(pathread).value.get() len_value = len(leader_value) leader_value.append(undefined) assert len(cfg_.forcepermissive.option(pathread).value.get()) == len_value with warnings.catch_warnings(record=True) as w: cfg2_.forcepermissive.option(pathread).value.set(leader_value) new_leader_value = cfg_.forcepermissive.option(pathread).value.get() len_new = len(new_leader_value) assert len_value + 1 == len_new assert new_leader_value[-1] == kwargs['default_multi'] follower_path = pathread.rsplit('.', 1)[0] if follower_path.endswith('val1') or follower_path.endswith('val2'): follower_path += '.third' + follower_path[-4:] else: follower_path += '.third' for idx in range(len_new): assert cfg_.forcepermissive.option(follower_path, idx).value.get() == kwargs['default_multi'] # if not submulti_: value = 'value' else: value = ['value'] leader_value.append(value) assert len(cfg_.forcepermissive.option(pathread).value.get()) == len(new_leader_value) cfg2_.forcepermissive.option(pathread).value.set(leader_value) assert cfg_.forcepermissive.option(pathread).value.get()[-1] == value @autocheck def autocheck_pop_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): isleader = cfg.unrestraint.option(pathread).option.isleader() submulti_ = cfg.unrestraint.option(pathread).option.issubmulti() if not isleader: return if confwrite is not None: cfg_ = cfg.config(confwrite) else: cfg_ = cfg if confread is not None: cfg2_ = cfg.config(confread) else: cfg2_ = cfg if not kwargs.get('propertyerror', False): if not submulti_: values = ['value1', 'value2', 'value3', 'value4'] follower_value = 'follower' else: values = [['value1'], ['value2'], ['value3'], ['value4']] follower_value = ['follower'] followers = [kwargs['default_multi'], follower_value, kwargs['default_multi'], kwargs['default_multi']] a_follower = pathwrite.rsplit('.', 1)[0] if a_follower.endswith('val1') or a_follower.endswith('val2'): a_follower += '.third' + a_follower[-4:] else: a_follower += '.third' with warnings.catch_warnings(record=True) as w: cfg_.forcepermissive.option(pathread).value.set(values) cfg_.forcepermissive.option(a_follower, 1).value.set(follower_value) cfg2_.forcepermissive.option(a_follower, 0).value.get() == kwargs['default_multi'] assert cfg2_.forcepermissive.option(a_follower, 0).owner.isdefault() is True cfg2_.forcepermissive.option(a_follower, 1).value.get() == follower_value assert cfg2_.forcepermissive.option(a_follower, 1).owner.isdefault() is False cfg2_.forcepermissive.option(a_follower, 2).value.get() == kwargs['default_multi'] assert cfg2_.forcepermissive.option(a_follower, 2).owner.isdefault() is True cfg2_.forcepermissive.option(a_follower, 3).value.get() == kwargs['default_multi'] assert cfg2_.forcepermissive.option(a_follower, 3).owner.isdefault() is True # cfg_.forcepermissive.option(pathread).value.pop(3) cfg2_.forcepermissive.option(a_follower, 0).value.get() == kwargs['default_multi'] assert cfg2_.forcepermissive.option(a_follower, 0).owner.isdefault() is True cfg2_.forcepermissive.option(a_follower, 1).value.get() == follower_value assert cfg2_.forcepermissive.option(a_follower, 1).owner.isdefault() is False cfg2_.forcepermissive.option(a_follower, 2).value.get() == kwargs['default_multi'] assert cfg2_.forcepermissive.option(a_follower, 2).owner.isdefault() is True # cfg_.forcepermissive.option(pathread).value.pop(0) cfg2_.forcepermissive.option(a_follower, 0).value.get() == follower_value assert cfg2_.forcepermissive.option(a_follower, 0).owner.isdefault() is False cfg2_.forcepermissive.option(a_follower, 1).value.get() == kwargs['default_multi'] assert cfg2_.forcepermissive.option(a_follower, 1).owner.isdefault() is True # cfg_.forcepermissive.option(pathread).value.pop(0) cfg2_.forcepermissive.option(a_follower, 0).value.get() == kwargs['default_multi'] assert cfg2_.forcepermissive.option(a_follower, 0).owner.isdefault() is True @autocheck def autocheck_reset_value_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): # check if is a multi, a leader or a follower isfollower = cfg.unrestraint.option(pathread).option.isfollower() _set_value(cfg, pathwrite, confwrite, **kwargs) # reset value with permissive with warnings.catch_warnings(record=True) as w: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): if not isfollower: if confwrite is not None: cfg_ = cfg.forcepermissive.config(confwrite) else: cfg_ = cfg.forcepermissive cfg_.option(pathwrite).value.reset() else: cfg.forcepermissive.option(pathwrite, 1).value.reset() elif kwargs.get('permissive', False): if not isfollower: if confwrite is not None: cfg_ = cfg.forcepermissive.config(confwrite) else: cfg_ = cfg.forcepermissive cfg_.option(pathwrite).value.reset() else: cfg.forcepermissive.option(pathwrite, 1).value.reset() #FIXME else: # if not isfollower: # raises(PropertiesOptionError, "cfg.forcepermissive.config(confwrite).option(pathwrite).value.reset()") # else: # raises(PropertiesOptionError, "cfg.forcepermissive.option(pathwrite, 1).value.reset()") _autocheck_default_value(cfg, pathread, confread, **kwargs) if confread != confwrite: _autocheck_default_value(cfg, pathwrite, confwrite, **kwargs) @autocheck def autocheck_display(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): """re set value """ if kwargs['callback']: return make_dict = kwargs['make_dict'] make_dict_value = kwargs['make_dict_value'] if confread is not None: cfg_ = cfg.config(confread) else: cfg_ = cfg if confwrite is not None: cfg2_ = cfg.config(confwrite) else: cfg2_ = cfg assert cfg_.value.dict() == make_dict if confread != confwrite: assert(cfg2_.value.dict()) == make_dict _set_value(cfg, pathwrite, confwrite, **kwargs) assert cfg_.value.dict() == make_dict_value if confread != confwrite: assert(cfg2_.value.dict()) == make_dict_value @autocheck def autocheck_property(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): """get property from path """ # check if is a multi or a follower multi = cfg.unrestraint.option(pathread).option.ismulti() isfollower = cfg.unrestraint.option(pathread).option.isfollower() default_props, properties = _getproperties(multi, isfollower, kwargs) if confread == confwrite: _check_properties(cfg, mcfg, pathread, confread, kwargs, default_props, default_props) else: _check_properties(cfg, mcfg, pathread, confwrite, kwargs, default_props, default_props) # set properties without permissive for prop in properties: if confwrite is not None: cfg_ = cfg.unrestraint.config(confwrite) else: cfg_ = cfg.unrestraint cfg_.option(pathwrite).property.add(prop) if confread == confwrite: _check_properties(cfg, mcfg, pathread, confread, kwargs, properties, properties) else: _check_properties(cfg, mcfg, pathread, confwrite, kwargs, properties, properties) @autocheck def autocheck_property_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): _property_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs) @autocheck def autocheck_reset_property(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): """check properties after set with permissive """ # check if is a multi or a follower multi = cfg.unrestraint.option(pathread).option.ismulti() isfollower = cfg.unrestraint.option(pathread).option.isfollower() default_props, properties = _getproperties(multi, isfollower, kwargs) _property_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs) # reset properties without permissive if confwrite is not None: cfg_ = cfg.unrestraint.config(confwrite) else: cfg_ = cfg.unrestraint cfg_.option(pathwrite).property.reset() if confread == confwrite: _check_properties(cfg, mcfg, pathread, confread, kwargs, default_props, default_props) else: _check_properties(cfg, mcfg, pathread, confwrite, kwargs, default_props, default_props) @autocheck def autocheck_reset_property_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): # check if is a multi or a follower multi = cfg.unrestraint.option(pathread).option.ismulti() isfollower = cfg.unrestraint.option(pathread).option.isfollower() default_props, properties = _getproperties(multi, isfollower, kwargs) _property_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs) for prop in properties: cfg.unrestraint.option(pathwrite).property.add(prop) cfg.unrestraint.option(pathwrite).property.reset() _check_properties(cfg, mcfg, pathread, confwrite, kwargs, default_props, default_props) @autocheck def autocheck_context_owner(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): owner = cfg.owner.get() assert owner == kwargs['owner'] @autocheck def autocheck_owner_with_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): """value is now changed, check owner in this case """ _set_value(cfg, pathwrite, confwrite, **kwargs) _check_owner(cfg, pathread, confwrite, kwargs, kwargs['owner'], kwargs['owner']) if confread != confwrite: _check_owner(cfg, pathread, confread, kwargs, kwargs['owner'], kwargs['owner']) @autocheck def autocheck_default_owner_with_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): isfollower = cfg.unrestraint.option(pathread).option.isfollower() _set_value(cfg, pathwrite, confwrite, **kwargs) if confwrite is not None: cfg_ = cfg.config(confread) cfg2_ = cfg.config(confwrite) else: cfg_ = cfg cfg2_ = cfg # test if is default owner without permissive if not isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg_.option(pathread).owner.isdefault() is False if confwrite != confread: assert cfg2_.option(pathread).owner.isdefault() is False #FIXME else: # raises(PropertiesOptionError, "cfg.config(confwrite).option(pathread).owner.isdefault()") # raises(PropertiesOptionError, "cfg.config(confread).option(pathread).owner.isdefault()") else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert cfg2_.option(pathread, 0).owner.isdefault() is True assert cfg2_.option(pathread, 1).owner.isdefault() is False if confwrite != confread: assert cfg_.option(pathread, 0).owner.isdefault() is True assert cfg_.option(pathread, 1).owner.isdefault() is False #FIXME else: # raises(PropertiesOptionError, "cfg.config(confwrite).option(pathread, 0).owner.isdefault()") # raises(PropertiesOptionError, "cfg.config(confread).option(pathread, 0).owner.isdefault()") @autocheck def autocheck_default_owner_with_value_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): # check if is a isfollower isfollower = cfg.unrestraint.option(pathread).option.isfollower() _set_value(cfg, pathwrite, confwrite, **kwargs) def do(conf): # test if is default owner with permissive if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg if not kwargs.get('propertyerror', False): if not isfollower: assert cfg_.forcepermissive.option(pathread).owner.isdefault() is False else: assert cfg_.forcepermissive.option(pathread, 0).owner.isdefault() is True assert cfg_.forcepermissive.option(pathread, 1).owner.isdefault() is False #FIXME else: # raises(PropertiesOptionError, "cfg.config(conf).forcepermissive.option(pathread).owner.isdefault()") do(confwrite) if confwrite != confread: do(confread) @autocheck def autocheck_set_owner_no_value(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): isfollower = cfg.unrestraint.option(pathread).option.isfollower() if confwrite is not None: cfg_ = cfg.forcepermissive.config(confwrite) else: cfg_ = cfg.forcepermissive if not kwargs.get('propertyerror', False): if not isfollower: raises(ConfigError, "cfg_.option(pathwrite).owner.set('new_user')") else: raises(ConfigError, "cfg_.option(pathwrite, 1).owner.set('new_user')") @autocheck def autocheck_set_owner(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): # test set owner without permissive isfollower = cfg.unrestraint.option(pathread).option.isfollower() _set_value(cfg, pathwrite, confwrite, **kwargs) if confwrite is not None: cfg_ = cfg.config(confwrite) else: cfg_ = cfg # set owner without permissive if not isfollower: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): cfg_.option(pathwrite).owner.set('new_user') raises(ValueError, "cfg_.option(pathwrite).owner.set('default')") raises(ValueError, "cfg_.option(pathwrite).owner.set('forced')") #FIXME else: # raises(PropertiesOptionError, "cfg.config(confwrite).option(pathwrite).owner.set('new_user')") else: if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): cfg.option(pathwrite, 1).owner.set('new_user') #FIXME else: # raises(PropertiesOptionError, "cfg.option(pathwrite, 1).owner.set('new_user')") _check_owner(cfg, pathread, confwrite, kwargs, owners.new_user, kwargs['owner']) if confwrite != confread: _check_owner(cfg, pathread, confread, kwargs, owners.new_user, owners.meta) @autocheck def autocheck_set_owner_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): isfollower = cfg.unrestraint.option(pathread).option.isfollower() _set_value(cfg, pathwrite, confwrite, **kwargs) if confwrite is not None: cfg_ = cfg.forcepermissive.config(confwrite) else: cfg_ = cfg.forcepermissive # set owner with permissive if not kwargs.get('propertyerror', False): if not isfollower: cfg_.option(pathwrite).owner.set('new_user1') else: cfg.forcepermissive.option(pathwrite, 1).owner.set('new_user1') #FIXME else: # if not isfollower: # raises(PropertiesOptionError, # "cfg.forcepermissive.config(confwrite).option(pathwrite).owner.set('new_user1')") # else: # raises(PropertiesOptionError, # "cfg.forcepermissive.option(pathwrite, 1).owner.set('new_user1')") _check_owner(cfg, pathread, confwrite, kwargs, 'new_user1', 'new_user1') if confwrite != confread: _check_owner(cfg, pathread, confread, kwargs, 'new_user1', 'new_user1') @autocheck def autocheck_option(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): expected_name = pathread.split('.')[-1] if not kwargs.get('permissive_od', False): current_name = cfg.option(pathread).option.name() assert current_name == cfg.forcepermissive.option(pathread).option.name() assert current_name == cfg.unrestraint.option(pathread).option.name() doc = cfg.option(pathread).option.doc() assert doc == cfg.forcepermissive.option(pathread).option.doc() assert doc == cfg.unrestraint.option(pathread).option.doc() elif not kwargs.get('propertyerror', False): raises(PropertiesOptionError, "cfg.option(pathread).option.name()") current_name = cfg.forcepermissive.option(pathread).option.name() assert current_name == cfg.unrestraint.option(pathread).option.name() raises(PropertiesOptionError, "cfg.option(pathread).option.doc()") doc = cfg.forcepermissive.option(pathread).option.doc() assert doc == cfg.unrestraint.option(pathread).option.doc() else: raises(PropertiesOptionError, "cfg.option(pathread).option.name()") raises(PropertiesOptionError, "cfg.forcepermissive.option(pathread).option.name()") current_name = cfg.unrestraint.option(pathread).option.name() raises(PropertiesOptionError, "cfg.option(pathread).option.doc()") raises(PropertiesOptionError, "cfg.forcepermissive.option(pathread).option.doc()") doc = cfg.unrestraint.option(pathread).option.doc() assert current_name == expected_name if expected_name.endswith('val1') or expected_name.endswith('val2'): expected_name = expected_name[:-4] if kwargs['symlink']: assert doc == "{}'s option link".format(expected_name) else: assert doc == "{}'s option".format(expected_name) @autocheck def autocheck_permissive(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): """test permissive for hidden and disabled value """ # no permissive before if confwrite is not None: cfg_ = cfg.unrestraint.config(confwrite) else: cfg_ = cfg.unrestraint if confread is not None: cfg2_ = cfg.config(confread).unrestraint else: cfg2_ = cfg.unrestraint assert cfg_.option(pathread).permissive.get() == frozenset() if kwargs.get('permissive_od', False): assert cfg_.option(pathread.rsplit('.', 1)[0]).permissive.get() == frozenset() # cannot access to hidden value without forcepermissive # and to disabled value (with forcepermissive too) # # with meta confread == confwrite _autocheck_default_value(cfg, pathread, confread, **kwargs) # set permissive cfg_.option(pathwrite).permissive.set(frozenset(['disabled'])) callback = kwargs['callback'] if callback: if pathread.endswith('val1') or pathread.endswith('val2'): call_path = pathread[:-4] + 'call' + pathread[-4:] else: call_path = pathread + 'call' cfg_.option(call_path).permissive.set(frozenset(['disabled'])) # have permissive? assert cfg_.option(pathread).permissive.get() == frozenset(['disabled']) #if confwrite != confread: # assert cfg.config(confread).unrestraint.option(pathread).permissive.get() == frozenset(['disabled']) # can access to disabled value ckwargs = copy(kwargs) ckwargs['propertyerror'] = False _autocheck_default_value(cfg, pathread, confwrite, **ckwargs) cfg2_.option(pathwrite).permissive.set(frozenset(['disabled', 'hidden'])) if kwargs['callback']: cfg2_.option(call_path).permissive.set(frozenset(['disabled', 'hidden'])) # can access to all value except when optiondescript have hidden if not ckwargs.get('permissive_od', False): ckwargs['permissive'] = False _autocheck_default_value(cfg, pathread, confread, **ckwargs) if ckwargs.get('permissive_od', False): # set permissive to OptionDescription cfg2_.option(pathwrite.rsplit('.', 1)[0]).permissive.set(frozenset(['disabled', 'hidden'])) ckwargs['permissive'] = False _autocheck_default_value(cfg, pathread, confread, **ckwargs) #if confread != confwrite: # _autocheck_default_value(cfg, pathread, confwrite, **ckwargs) # only hidden cfg2_.option(pathwrite).permissive.set(frozenset(['hidden'])) if callback: cfg2_.option(call_path).permissive.set(frozenset(['hidden'])) if ckwargs.get('permissive_od', False): _autocheck_default_value(cfg, pathread, confread, **ckwargs) cfg2_.option(pathwrite.rsplit('.', 1)[0]).permissive.set(frozenset(['hidden'])) ckwargs = copy(kwargs) ckwargs['permissive'] = False _autocheck_default_value(cfg, pathread, confread, **ckwargs) # no permissive cfg2_.option(pathwrite).permissive.set(frozenset()) if callback: cfg2_.option(call_path).permissive.set(frozenset()) if ckwargs.get('permissive_od', False): _autocheck_default_value(cfg, pathread, confread, **ckwargs) cfg2_.option(pathwrite.rsplit('.', 1)[0]).permissive.set(frozenset()) _autocheck_default_value(cfg, pathread, confread, **kwargs) @autocheck def autocheck_option_get(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): if '.' in pathread: name = pathread.rsplit('.', 1)[1] else: name = pathread assert cfg.unrestraint.option(pathread).option.name() == name @autocheck def autocheck_find(cfg, mcfg, pathread, pathwrite, confread, confwrite, **kwargs): def _getoption(opt): opt = opt.option.get() if opt.impl_is_dynsymlinkoption(): opt = opt.opt return opt def _getoptions(opts): nopts = [] for opt in opts: nopts.append(_getoption(opt)) return nopts if '.' in pathread: name = pathread.rsplit('.', 1)[1] else: name = pathread option = _getoption(cfg.unrestraint.option(pathread)) def do(conf): if conf is not None: cfg_ = cfg.config(conf) else: cfg_ = cfg if not kwargs.get('permissive', False) and not kwargs.get('propertyerror', False): assert option == _getoption(cfg_.option.find(name, first=True)) assert option == _getoption(cfg_.forcepermissive.option.find(name, first=True)) elif kwargs.get('permissive', False): raises(AttributeError, "cfg_.option.find(name, first=True)") assert option == _getoption(cfg_.forcepermissive.option.find(name, first=True)) else: raises(AttributeError, "cfg_.option.find(name, first=True)") raises(AttributeError, "cfg_.forcepermissive.option.find(name, first=True)") assert option == _getoption(cfg_.unrestraint.option.find(name, first=True)) assert [option] == _getoptions(cfg_.unrestraint.option.find(name)) do(confread) if confread != confwrite: do(confwrite) def check_all(cfg, paths_, path, meta, multi, default, default_multi, require, consistency, callback, symlink, weakrefs, **kwargs): def _build_make_dict(): dico = {} dico_value = {} if multi is False: value = FIRST_VALUE elif multi is True: value = LIST_FIRST_VALUE else: value = SUBLIST_FIRST_VALUE if not default or multi is submulti: if not multi: default_value = None else: default_value = [] else: default_value = value kwargs['default'] = default_value is_dyn = False is_leader = False dyns = [] has_value = False for cpath, options in paths_.items(): if options is None: break if '.' in cpath: dirname, name = cpath.split('.')[-2:] for dname in cpath.split('.')[:-1]: if options.get(dname, {}).get('dyn'): is_dyn = True if options.get(dname, {}).get('leader'): is_leader = True else: dirname = '' name = cpath if options.get(dirname, {}).get('hidden'): continue allow_req = require and req no_propertieserror = not options.get(name, {}).get('disabled') and not options.get(name, {}).get('hidden') if is_dyn: dyns.append(no_propertieserror or allow_req) if not is_dyn and (no_propertieserror or allow_req): dico[cpath] = default_value if symlink: dico[cpath + 'link'] = default_value if path == cpath: dico_value[cpath] = value if symlink: dico_value[cpath + 'link'] = value else: dico_value[cpath] = default_value if symlink: dico_value[cpath + 'link'] = default_value has_value = True isfollower = False if '.' in path and is_leader and not path.rsplit('.', 1)[1].startswith('first'): isfollower = True if not multi is submulti: kwargs['default'] = None if is_dyn and dyns: idx = 0 for cpath in list(paths_.keys())[len(dyns):]: if dyns[idx]: dico[cpath] = default_value if symlink: dico[cpath + 'link'] = default_value if path == cpath: dico_value[cpath] = value if symlink: dico_value[cpath + 'link'] = value else: dico_value[cpath] = default_value if symlink: dico_value[cpath + 'link'] = default_value idx += 1 if idx == len(dyns): idx = 0 if require: if not req: dico['extraoptrequire'] = None dico_value['extraoptrequire'] = None if symlink: dico['extraoptrequirelink'] = None dico_value['extraoptrequirelink'] = None else: dico['extraoptrequire'] = 'value' dico_value['extraoptrequire'] = 'value' if symlink: dico['extraoptrequirelink'] = 'value' dico_value['extraoptrequirelink'] = 'value' if consistency and has_value: cpath = list(dico.keys())[0] if "." in cpath: cpath = cpath.rsplit('.', 1)[0] + '.' else: cpath = '' if multi: value = [] else: value = None if is_dyn: dico[cpath + 'extraoptconsistencyval1'] = value dico_value[cpath + 'extraoptconsistencyval1'] = value if is_leader: spath = cpath.split('.') spath[-2] = spath[-2][:-1] + '2' spath[-3] = spath[-3][:-1] + '2' npath = '.'.join(spath) + 'extraoptconsistencyval2' else: npath = cpath[:-2] + '2.' + 'extraoptconsistencyval2' dico[npath] = value dico_value[npath] = value else: dico[cpath + 'extraoptconsistency'] = value dico_value[cpath + 'extraoptconsistency'] = value if is_leader: for cpath in list(paths_.keys())[len(dyns):]: if cpath.endswith('.first') or cpath.endswith('.firstval1') or cpath.endswith('.firstval2'): second_path = cpath.rsplit('.', 1)[0] + '.second' third_path = cpath.rsplit('.', 1)[0] + '.third' cons_path = cpath.rsplit('.', 1)[0] + '.extraoptconsistency' if is_dyn: suffix = cpath[-4:] second_path += suffix third_path += suffix cons_path += suffix # if default_multi: if multi is not submulti: dvalue = SECOND_VALUE else: dvalue = LIST_SECOND_VALUE else: dvalue = [] if dvalue == [] and multi is not submulti: dvalue = None # kwargs['default_multi'] = dvalue if isfollower: kwargs['default'] = dvalue len_leader = len(dico[cpath]) if second_path in dico: dico[second_path] = [dvalue] * len_leader if symlink: dico[second_path + 'link'] = [dvalue] * len_leader if third_path in dico: dico[third_path] = [dvalue] * len_leader if symlink: dico[third_path + 'link'] = [dvalue] * len_leader if cons_path in dico: dico[cons_path] = [dvalue] * len_leader # len_leader = len(dico_value[cpath]) if second_path in dico_value: dico_value[second_path] = [dvalue] * len_leader if symlink: dico_value[second_path + 'link'] = [dvalue] * len_leader if third_path in dico_value: dico_value[third_path] = [dvalue] * len_leader if symlink: dico_value[third_path + 'link'] = [dvalue] * len_leader if cons_path in dico_value: dico_value[cons_path] = [dvalue] * len_leader return is_dyn, dico, dico_value if DISPLAY: text = u' {} launch tests for {}'.format(ICON, path) if multi is True: text += u' as a multi' elif multi is submulti: text += u' as a submulti' if default is True: text += u' with default' if multi is True: text += u' with default value' if default_multi is True: text += u' with default multi' if require: text += u' with requirement' if consistency: text += u' with consistency' text += u', kwargs: {}'.format(kwargs) print(text) if not require: requires = [False] else: requires = [False, True] confwrite = confread = None idx = 0 for req in requires: is_dyn, kwargs['make_dict'], kwargs['make_dict_value'] = _build_make_dict() kwargs['callback'] = callback kwargs['symlink'] = symlink for func in autocheck_registers: cfg_name = 'conftest' + str(idx) idx += 1 ncfg = cfg.config.copy(session_id=cfg_name) if meta: confwrite = None confread = cfg_name mcfg = MetaConfig([ncfg], session_id='metatest') weakrefs.append(weakref.ref(cfg)) else: mcfg = ncfg ckwargs = copy(kwargs) if meta: mcfg.owner.set('meta') ckwargs['owner'] = owners.meta else: ckwargs['owner'] = OWNER if mcfg.unrestraint.option(path).option.isfollower(): dirname = path.rsplit('.', 1)[0] leader_path = dirname + '.first' leader_path_2 = None if dirname.endswith('val1') or dirname.endswith('val2'): leader_path += 'val1' leader_path = leader_path.replace('val2', 'val1') leader_path_2 = leader_path.replace('val1', 'val2') if multi is submulti: value = SUBLIST_SECOND_VALUE else: value = LIST_SECOND_VALUE with warnings.catch_warnings(record=True) as w: mcfg.option(leader_path).value.set(value) ckwargs['make_dict'][leader_path] = value ckwargs['make_dict_value'][leader_path] = value if symlink: ckwargs['make_dict'][leader_path + 'link'] = value ckwargs['make_dict_value'][leader_path + 'link'] = value if leader_path_2: with warnings.catch_warnings(record=True) as w: mcfg.option(leader_path_2).value.set(value) ckwargs['make_dict'][leader_path_2] = value ckwargs['make_dict_value'][leader_path_2] = value if symlink: ckwargs['make_dict'][leader_path_2 + 'link'] = value ckwargs['make_dict_value'][leader_path_2 + 'link'] = value if default_multi: if multi is not submulti: dvalue = SECOND_VALUE else: dvalue = LIST_SECOND_VALUE elif multi is submulti: dvalue = [] else: dvalue = None def do(suffix, oldsuffix=None): if suffix: ldirname = dirname.replace(oldsuffix, suffix) else: ldirname = dirname npath = ldirname + '.second' + suffix if npath in ckwargs['make_dict']: ckwargs['make_dict'][npath] = [dvalue] * len(value) ckwargs['make_dict_value'][npath] = [dvalue] * len(value) if symlink: ckwargs['make_dict'][npath + 'link'] = [dvalue] * len(value) ckwargs['make_dict_value'][npath + 'link'] = [dvalue] * len(value) if path == npath: ckwargs['make_dict_value'][npath][-1] = ckwargs['make_dict_value'][leader_path][-1] if symlink: ckwargs['make_dict_value'][npath + 'link'][-1] = ckwargs['make_dict_value'][leader_path][-1] npath = ldirname + '.third' + suffix if npath in ckwargs['make_dict']: ckwargs['make_dict'][npath] = [dvalue] * len(value) ckwargs['make_dict_value'][npath] = [dvalue] * len(value) if symlink: ckwargs['make_dict'][npath + 'link'] = [dvalue] * len(value) ckwargs['make_dict_value'][npath + 'link'] = [dvalue] * len(value) if path == npath: ckwargs['make_dict_value'][npath][-1] = ckwargs['make_dict_value'][leader_path][-1] if symlink: ckwargs['make_dict_value'][npath + 'link'][-1] = ckwargs['make_dict_value'][leader_path][-1] npath = ldirname + '.extraoptconsistency' + suffix if npath in ckwargs['make_dict']: ckwargs['make_dict'][npath] = [dvalue] * len(value) ckwargs['make_dict_value'][npath] = [dvalue] * len(value) if symlink: ckwargs['make_dict'][npath + 'link'] = [dvalue] * len(value) ckwargs['make_dict_value'][npath + 'link'] = [dvalue] * len(value) if not is_dyn: do('') else: #do(dirname[-4:]) do('val1', 'val2') do('val2', 'val1') ncfg.property.read_write() mcfg.property.read_write() if req: name = 'extraoptrequire' if symlink: name += 'link' mcfg.option(name).value.set('value') if 'permissive' in ckwargs and not 'permissive_od' in ckwargs or \ 'propertyerror' in ckwargs and not 'propertyerror_od' in ckwargs: for to_del in ['permissive', 'propertyerror', 'extra_properties']: if to_del in ckwargs: del ckwargs[to_del] if DISPLAY: print(u' {} {}'.format(ICON, func.__name__)) pathread = path if symlink: pathwrite = path + 'link' else: pathwrite = path try: func(mcfg, ncfg, pathread, pathwrite, confread, confwrite, **ckwargs) except Exception as err: msg = u'error in function {} for {}'.format(func.__name__, path) if multi is True: msg += u' as a multi' elif multi is submulti: msg += u' as a submulti' if default is True: msg += u' with default value' if callback is True: msg += u' (callback)' if symlink is True: msg += u' (symlink)' print(u'{}: {}'.format(msg, ckwargs)) raise err if meta: del mcfg del ncfg def check_deref(weakrefs): """try if all elements are dereferenced """ for wrf in weakrefs: assert wrf() is None def make_conf(options, multi, default, default_multi, require, consistency, callback, symlink): weakrefs = [] dyn = [] goptions = [] def make_option(path, option_infos, in_leader, leader): option_type = 'str' option_properties = [] option_requires = [] isfollower = False if in_leader and symlink: return None, None, None if option_infos is not None: if require: return None, None, None for prop in PROPERTIES: if option_infos.get(prop, False) is True: if not require: option_properties.append(prop) else: option_requires.append({'option': goptions[0], 'expected': None, 'action': prop}) isfollower = option_infos.get('follower', False) args = [path, "{}'s option".format(path)] kwargs = {} call_kwargs = {} if option_properties != []: kwargs['properties'] = tuple(option_properties) if callback: call_kwargs['properties'] = tuple(option_properties) if option_requires != []: if callback: call_kwargs['requires'] = option_requires else: kwargs['requires'] = option_requires if multi and path is not 'extraoptrequire': kwargs['multi'] = multi if callback: call_kwargs['multi'] = multi if ((not in_leader or leader) and default) and path is not 'extraoptrequire' and not path.endswith('extraoptconsistency'): if multi is False: value = FIRST_VALUE elif multi is True: value = LIST_FIRST_VALUE else: value = SUBLIST_EMPTY_VALUE if callback: kwargs['callback'] = return_str call_kwargs['default'] = value else: kwargs['default'] = value elif callback: return None, None, None if default_multi and path is not 'extraoptrequire': if multi is not submulti: value = SECOND_VALUE else: value = LIST_SECOND_VALUE kwargs['default_multi'] = value tiramisu_option = OPTIONS_TYPE[option_type]['option'] if callback: largs = [path + 'call', "{}'s callback option".format(path)] objcall = tiramisu_option(*largs, **call_kwargs) kwargs['callback_params'] = Params(ParamOption(objcall)) else: objcall = None if symlink and not path.endswith('extraoptconsistency'): sobj = tiramisu_option(args[0] + 'link', args[1] + ' link', **kwargs) kwargs = {} args[1] = sobj tiramisu_option = SymLinkOption else: sobj = None obj = tiramisu_option(*args, **kwargs) if not 'extraopt' in path and consistency: if require: if symlink: gopt = goptions[2] else: gopt = goptions[1] else: gopt = goptions[0] obj.impl_add_consistency('not_equal', gopt, warnings_only=True, transitive=False) return obj, objcall, sobj def make_optiondescriptions(path, collected): infos = collected.get('properties', {}) properties = [] kwargs = {} optiondescription = OptionDescription for prop in PROPERTIES: if infos.get(prop, False) is True: properties.append(prop) if infos.get('leader', False) is True: if not multi: return optiondescription = Leadership if infos.get('dyn', False) is True: if symlink: return optiondescription = DynOptionDescription kwargs['callback'] = return_list dyn.append(path) options = [] if 'options' in collected: options.extend(collected['options']) for key, values in collected.items(): if key in ['options', 'properties']: continue option = make_optiondescriptions(key, values) if option is None: return options.append(option) if properties != []: kwargs['properties'] = tuple(properties) obj = optiondescription(path, "{}'s optiondescription".format(path), options, **kwargs) weakrefs.append(weakref.ref(obj)) return obj collect_options = {} if require or consistency: noptions = OrderedDict() if require: noptions['extraoptrequire'] = {} if consistency: subpath = list(options.keys())[0] if '.' in subpath: subpath = subpath.rsplit('.', 1)[0] + '.' else: subpath = '' noptions[subpath + 'extraoptconsistency'] = {} noptions.update(options) else: noptions = options for path, option in noptions.items(): if option is None: continue local_collect_options = collect_options for optiondescription in path.split('.')[:-1]: local_collect_options.setdefault(optiondescription, {'properties': {}}) local_collect_options = local_collect_options[optiondescription] local_collect_options['properties'].update(option.get(optiondescription, {})) option_name = path.split('.')[-1] in_leader = False if '.' in path: name_od = path.rsplit('.', 1)[0] if '.' in name_od: subod, name_od = name_od.split('.') oddescr = collect_options.get(subod, {}) else: oddescr = collect_options in_leader = oddescr.get(name_od, {}).get('properties', {}).get('leader') leader = in_leader and path.endswith('first') obj, objcall, sobj = make_option(option_name, option.get(option_name), in_leader, leader) if obj is None: return None, None, None weakrefs.append(weakref.ref(obj)) if callback: weakrefs.append(weakref.ref(objcall)) if sobj is not None: weakrefs.append(weakref.ref(sobj)) if '.' in path: if leader: local_collect_options.setdefault('options', []).insert(0, obj) else: local_collect_options.setdefault('options', []).append(obj) else: local_collect_options.setdefault('options', []).append(obj) goptions.append(obj) if callback: local_collect_options.setdefault('options', []).append(objcall) goptions.append(objcall) if sobj is not None: local_collect_options.setdefault('options', []).append(sobj) goptions.append(sobj) rootod = make_optiondescriptions('root', collect_options) if rootod is None: return None, None, None cfg = Config(rootod, session_id='conftest') weakrefs.append(weakref.ref(cfg)) del goptions return cfg, weakrefs, dyn DICT_PATHS = [ # test a config without optiondescription OrderedDict([('first', {}), ('second', {'second': {'disabled': True}}), ('third', {'third': {'hidden': True}}) ]), # test a config with two optiondescription OrderedDict([('subod.subsubod.first', {}), ('subod.subsubod.second', {'second': {'disabled': True}}), ('subod.subsubod.third', {'third': {'hidden': True}})]), # test a config with leadership OrderedDict([('odleader.first', {'odleader': {'leader': True}}), ('odleader.second', {'odleader': {'leader': True}, 'second': {'disabled': True, 'follower': True}}), ('odleader.third', {'odleader': {'leader': True}, 'third': {'hidden': True, 'follower': True}})]), # test a config with dynoption OrderedDict([('subod.first', {'subod': {'dyn': True}}), ('subod.second', {'second': {'disabled': True}}), ('subod.third', {'third': {'hidden': True}}), ('subodval1.firstval1', None), ('subodval1.secondval1', None), ('subodval1.thirdval1', None), ('subodval2.firstval2', None), ('subodval2.secondval2', None), ('subodval2.thirdval2', None)]), # test a config with dynoption subdir OrderedDict([('subod.subsubod.first', {'subsubod': {'dyn': True}}), ('subod.subsubod.second', {'subsubod': {'dyn': True}, 'second': {'disabled': True}}), ('subod.subsubod.third', {'subsubod': {'dyn': True}, 'third': {'hidden': True}}), ('subod.subsubodval1.firstval1', None), ('subod.subsubodval1.secondval1', None), ('subod.subsubodval1.thirdval1', None), ('subod.subsubodval2.firstval2', None), ('subod.subsubodval2.secondval2', None), ('subod.subsubodval2.thirdval2', None)]), # test a config with hidden subsubod OrderedDict([('subod.subsubod.first', {'subsubod': {'hidden': True}}), ('subod.subsubod.second', {'subsubod': {'hidden': True}}), ('subod.subsubod.third', {'subsubod': {'hidden': True}})]), # test a config with hidden dyn subsubod OrderedDict([('subod.subsubod.first', {'subsubod': {'dyn': True, 'hidden': True}}), ('subod.subsubod.second', {'subsubod': {'dyn': True, 'hidden': True}}), ('subod.subsubod.third', {'subsubod': {'dyn': True, 'hidden': True}}), ('subod.subsubodval1.firstval1', None), ('subod.subsubodval1.secondval1', None), ('subod.subsubodval1.thirdval1', None), ('subod.subsubodval2.firstval2', None), ('subod.subsubodval2.secondval2', None), ('subod.subsubodval2.thirdval2', None)]), # test a config with dyn subsubod with leadership OrderedDict([('subod.subsubod.first', {'subod': {'dyn': True}, 'subsubod': {'leader': True}}), ('subod.subsubod.second', {'subod': {'dyn': True}, 'subsubod' : {'leader': True}, 'second': {'disabled': True, 'follower': True}}), ('subod.subsubod.third', {'subod': {'dyn': True}, 'subsubod': {'leader': True}, 'third': {'hidden': True, 'follower': True}}), ('subodval1.subsubodval1.firstval1', None), ('subodval1.subsubodval1.secondval1', None), ('subodval1.subsubodval1.thirdval1', None), ('subodval2.subsubodval2.firstval2', None), ('subodval2.subsubodval2.secondval2', None), ('subodval2.subsubodval2.thirdval2', None)]), ] @pytest.fixture(scope="function", params=DICT_PATHS) def paths(request): if DISPLAY: print(u'\n{} {}: {}'.format(ICON, request.function.__name__, request.param)) return request.param def test_options(paths): def get_kwargs_option(options, kwargs, od=False): if options.get('mandatory', False): kwargs['mandatory'] = True if options.get('hidden', False) is True: kwargs['permissive'] = True if not od: kwargs.setdefault('extra_properties', []).append('hidden') else: kwargs['permissive_od'] = True if options.get('disabled', False) is True: kwargs['propertyerror'] = True if not od: kwargs.setdefault('extra_properties', []).append('disabled') else: kwargs['propertyerror_od'] = True def get_kwargs(path): kwargs = {} spath = path.split('.') get_kwargs_option(paths[path].get(spath[-1], {}), kwargs) if len(spath) > 1: get_kwargs_option(paths[path].get(spath[-2], {}), kwargs, od=True) return kwargs lpaths = list(paths.keys()) for meta in (False, True): for callback in (False, True): for consistency in (False, True): for require in (False, True): for default_multi in (False, True): for symlink in (False, True): if callback and default_multi: continue for default in (False, True): for multi in (False, True, submulti): pass # for meta in (True,): # for callback in (False,): # for consistency in (True,): # for require in (True,): # for default_multi in (True,): # for symlink in (False,): # if callback and default_multi: # continue # for default in (True,): # for multi in (submulti,): if multi is submulti and default: continue if multi is submulti and consistency: continue if multi is False and default_multi: continue cfg, weakrefs, dyn = make_conf(paths, multi, default, default_multi, require, consistency, callback, symlink) if cfg is None: continue if dyn: cnt = 0 idx = 0 for index, lpath in enumerate(lpaths): if paths[lpath]: cnt += 1 else: check_all(cfg, paths, lpaths[index], meta, multi, default, default_multi, require, consistency, callback, symlink, weakrefs, **get_kwargs(lpaths[idx])) idx += 1 if idx == cnt: idx = 0 else: for lpath in lpaths: check_all(cfg, paths, lpath, meta, multi, default, default_multi, require, consistency, callback, symlink, weakrefs, **get_kwargs(lpath)) del cfg check_deref(weakrefs)