from .autopath import do_autopath do_autopath() import pytest import warnings from tiramisu import Config from tiramisu.config import KernelConfig from tiramisu.setting import groups, owners from tiramisu import ChoiceOption, BoolOption, IntOption, FloatOption, \ StrOption, OptionDescription, SymLinkOption, IPOption, NetmaskOption, Leadership, \ undefined, Calculation, Params, ParamOption, ParamValue, ParamIndex, calc_value, \ valid_ip_netmask, ParamSelfOption from tiramisu.error import PropertiesOptionError, ConflictError, LeadershipError, ConfigError from tiramisu.i18n import _ from tiramisu.storage import list_sessions from .config import config_type, get_config, event_loop def return_val(): return 'val' def return_concat(*args): return '.'.join(list(args)) def return_list(value=None): return ['val', 'val'] def return_list2(*args): l = [] for arg in args: if isinstance(arg, list): l.extend(arg) else: l.append(arg) return l def return_value(value=None): return value def return_value2(*args, **kwargs): value = list(args) value.extend(kwargs.values()) return value def return_value3(value=None, index=None): if index is not None and isinstance(value, list): if len(value) > index: return value[index] return None return value def return_index(val1, val2=None, index=None, self=None): if index is None: return [val1, val2] if index == 0: return val1 if index == 1: return val2 def return_calc(i, j, k): return i + j + k def is_config(config, **kwargs): if isinstance(config, KernelConfig): return 'yes' return 'no' def return_raise(*arg): raise Exception('test') def return_valueerror(*arg): raise ValueError('test') def make_description_duplicates(): gcoption = ChoiceOption('name', 'GC name', ('ref', 'framework'), 'ref') ## dummy 1 gcdummy = BoolOption('dummy', 'dummy', default=False) objspaceoption = ChoiceOption('objspace', 'Object space', ('std', 'thunk'), 'std') booloption = BoolOption('bool', 'Test boolean option', default=True) intoption = IntOption('int', 'Test int option', default=0) floatoption = FloatOption('float', 'Test float option', default=2.3) stroption = StrOption('str', 'Test string option', default="abc") boolop = BoolOption('boolop', 'Test boolean option op', default=True) wantref_option = BoolOption('wantref', 'Test requires', default=False) wantframework_option = BoolOption('wantframework', 'Test requires', default=False) # dummy2 (same path) gcdummy2 = BoolOption('dummy', 'dummy2', default=True) # dummy3 (same name) gcdummy3 = BoolOption('dummy', 'dummy2', default=True) gcgroup = OptionDescription('gc', '', [gcoption, gcdummy, gcdummy2, floatoption]) descr = OptionDescription('constraints', '', [gcgroup, booloption, objspaceoption, wantref_option, stroption, wantframework_option, intoption, boolop, gcdummy3]) return descr @pytest.mark.asyncio async def test_identical_paths(): """If in the schema (the option description) there is something that have the same name, an exection is raised """ with warnings.catch_warnings(record=True) as w: with pytest.raises(ConflictError): make_description_duplicates() @pytest.mark.asyncio async def test_hidden_if_in2(config_type): intoption = IntOption('int', 'Test int option', default=0) hidden_property = Calculation(calc_value, Params(ParamValue('hidden'), kwargs={'condition': ParamOption(intoption), 'expected': ParamValue(1), 'default': ParamValue(None)})) stroption = StrOption('str', 'Test string option', default="abc", properties=(hidden_property,)) descr = OptionDescription('constraints', '', [stroption, intoption]) async with await Config(descr) as cfg_ori: await cfg_ori.property.read_write() cfg = await get_config(cfg_ori, config_type) assert not 'hidden' in await cfg.option('str').property.get() await cfg.option('int').value.set(1) with pytest.raises(PropertiesOptionError): await cfg.option('str').value.get() with pytest.raises(PropertiesOptionError): await cfg.option('str').value.set('uvw') if config_type == 'tiramisu-api': await cfg.send() assert 'hidden' in await cfg_ori.unrestraint.option('str').property.get() assert not await list_sessions() @pytest.mark.asyncio async def test_hidden_if_in_with_group(config_type): gcoption = ChoiceOption('name', 'GC name', ('ref', 'framework'), 'ref') gcdummy = BoolOption('dummy', 'dummy', default=False) floatoption = FloatOption('float', 'Test float option', default=2.3) objspaceoption = ChoiceOption('objspace', 'Object space', ('std', 'thunk'), 'std') booloption = BoolOption('bool', 'Test boolean option', default=True) intoption = IntOption('int', 'Test int option', default=0) stroption = StrOption('str', 'Test string option', default="abc") hidden_property = Calculation(calc_value, Params(ParamValue('hidden'), kwargs={'condition': ParamOption(intoption), 'expected': ParamValue(1), 'default': ParamValue(None)})) gcgroup = OptionDescription('gc', '', [gcoption, gcdummy, floatoption], properties=(hidden_property,)) descr = OptionDescription('constraints', '', [gcgroup, booloption, objspaceoption, stroption, intoption]) async with await Config(descr) as cfg_ori: await cfg_ori.property.read_write() cfg = await get_config(cfg_ori, config_type) assert not 'hidden' in await cfg_ori.option('str').property.get() await cfg.option('int').value.set(1) if config_type == 'tiramisu-api': await cfg.send() with pytest.raises(PropertiesOptionError): await cfg_ori.option('gc.name').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_disabled_with_group(): gcoption = ChoiceOption('name', 'GC name', ('ref', 'framework'), 'ref') gcdummy = BoolOption('dummy', 'dummy', default=False) floatoption = FloatOption('float', 'Test float option', default=2.3) objspaceoption = ChoiceOption('objspace', 'Object space', ('std', 'thunk'), 'std') booloption = BoolOption('bool', 'Test boolean option', default=True) intoption = IntOption('int', 'Test int option', default=0) stroption = StrOption('str', 'Test string option', default="abc") disabled_property = Calculation(calc_value, Params(ParamValue('disabled'), kwargs={'condition': ParamOption(intoption), 'expected': ParamValue(1), 'default': ParamValue(None)})) gcgroup = OptionDescription('gc', '', [gcoption, gcdummy, floatoption], properties=(disabled_property,)) descr = OptionDescription('constraints', '', [gcgroup, booloption, objspaceoption, stroption, intoption]) async with await Config(descr) as cfg: await cfg.property.read_write() assert await cfg.option('gc.name').value.get() await cfg.option('int').value.set(1) with pytest.raises(PropertiesOptionError): await cfg.option('gc.name').value.get() assert not await list_sessions() #____________________________________________________________ def make_description_callback(): gcoption = ChoiceOption('name', 'GC name', ('ref', 'framework'), 'ref') gcdummy = BoolOption('dummy', 'dummy') objspaceoption = ChoiceOption('objspace', 'Object space', ('std', 'thunk'), 'std') booloption = BoolOption('bool', 'Test boolean option', default=True) intoption = IntOption('int', 'Test int option', default=0) floatoption = FloatOption('float', 'Test float option', default=2.3) stroption = StrOption('str', 'Test string option', default="abc") boolop = BoolOption('boolop', 'Test boolean option op', default=True) hidden_property = Calculation(calc_value, Params(ParamValue('hidden'), kwargs={'condition': ParamOption(boolop), 'expected': ParamValue(True), 'default': ParamValue(None)})) wantref_option = BoolOption('wantref', 'Test requires', default=False, properties=(hidden_property,)) wantframework_option = BoolOption('wantframework', 'Test requires', default=False, properties=(hidden_property,)) gcgroup = OptionDescription('gc', '', [gcoption, gcdummy, floatoption]) descr = OptionDescription('constraints', '', [gcgroup, booloption, objspaceoption, wantref_option, stroption, wantframework_option, intoption, boolop]) return descr @pytest.mark.asyncio async def test_has_callback(): descr = make_description_callback() # here the owner is 'default' async with await Config(descr) as cfg: await cfg.property.read_write() await cfg.option('bool').value.set(False) # because dummy has a callback await cfg.property.add('freeze') await cfg.option('gc.dummy').property.add('frozen') with pytest.raises(PropertiesOptionError): await cfg.option('gc.dummy').value.set(True) assert not await list_sessions() @pytest.mark.asyncio async def test_freeze_and_has_callback(): descr = make_description_callback() async with await Config(descr) as cfg: await cfg.property.read_write() await cfg.option('bool').value.set(False) await cfg.property.add('freeze') await cfg.option('gc.dummy').property.add('frozen') with pytest.raises(PropertiesOptionError): await cfg.option('gc.dummy').value.set(True) assert not await list_sessions() @pytest.mark.asyncio async def test_callback(config_type): val1 = StrOption('val1', "", Calculation(return_val)) val2 = StrOption('val2', "") maconfig = OptionDescription('rootconfig', '', [val1, val2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == 'val' await cfg.option('val1').value.set('new-val') assert await cfg.option('val1').value.get() == 'new-val' await cfg.option('val1').value.reset() assert await cfg.option('val1').value.get() == 'val' assert not await list_sessions() @pytest.mark.asyncio async def test_params(): with pytest.raises(ValueError): Params('str') with pytest.raises(ValueError): Params(('str',)) with pytest.raises(ValueError): Params(kwargs={'a': 'str'}) @pytest.mark.asyncio async def test_param_option(): val1 = StrOption('val1', "") with pytest.raises(ValueError): ParamOption('str') with pytest.raises(AssertionError): ParamOption(val1, 'str') @pytest.mark.asyncio async def test_callback_with_error(config_type): val1 = StrOption("val1", "", Calculation(is_config, Params(ParamValue('string'), kwargs={'value': ParamValue('string')}))) maconfig = OptionDescription('rootconfig', '', [val1]) async with await Config(maconfig) as cfg: cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == 'no' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_value(config_type): val1 = StrOption('val1', "", 'val') val2 = StrOption('val2', "", Calculation(return_value, Params(ParamOption(val1)))) val3 = StrOption('val3', "", Calculation(return_value, Params(ParamValue('yes')))) val4 = StrOption('val4', "", Calculation(return_value, Params(kwargs={'value': ParamOption(val1)}))) val5 = StrOption('val5', "", Calculation(return_value, Params(ParamValue('yes')))) maconfig = OptionDescription('rootconfig', '', [val1, val2, val3, val4, val5]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == 'val' assert await cfg.option('val2').value.get() == 'val' assert await cfg.option('val4').value.get() == 'val' await cfg.option('val1').value.set('new-val') assert await cfg.option('val1').value.get() == 'new-val' assert await cfg.option('val2').value.get() == 'new-val' assert await cfg.option('val4').value.get() == 'new-val' await cfg.option('val1').value.reset() assert await cfg.option('val1').value.get() == 'val' assert await cfg.option('val2').value.get() == 'val' assert await cfg.option('val3').value.get() == 'yes' assert await cfg.option('val4').value.get() == 'val' assert await cfg.option('val5').value.get() == 'yes' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_value_tuple(config_type): val1 = StrOption('val1', "", 'val1') val2 = StrOption('val2', "", 'val2') val3 = StrOption('val3', "", Calculation(return_concat, Params((ParamOption(val1), ParamOption(val2))))) val4 = StrOption('val4', "", Calculation(return_concat, Params((ParamValue('yes'), ParamValue('no'))))) maconfig = OptionDescription('rootconfig', '', [val1, val2, val3, val4]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == 'val1' assert await cfg.option('val2').value.get() == 'val2' assert await cfg.option('val3').value.get() == 'val1.val2' assert await cfg.option('val4').value.get() == 'yes.no' await cfg.option('val1').value.set('new-val') assert await cfg.option('val3').value.get() == 'new-val.val2' await cfg.option('val1').value.reset() assert await cfg.option('val3').value.get() == 'val1.val2' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_value_force_permissive2(config_type): val1 = StrOption('val1', "", 'val', properties=('disabled',)) val2 = StrOption('val2', "", Calculation(return_value, Params(ParamOption(val1)))) val3 = StrOption('val3', "", Calculation(return_value, Params(ParamOption(val1, True)))) maconfig = OptionDescription('rootconfig', '', [val1, val2, val3]) async with await Config(maconfig) as cfg: await cfg.property.read_only() if config_type != 'tiramisu-api': with pytest.raises(ConfigError): await cfg.option('val2').value.get() await cfg.option('val3').value.get() is None else: with pytest.raises(ConfigError): await get_config(cfg, config_type) assert not await list_sessions() @pytest.mark.asyncio async def test_callback_value_force_permissive_kwargs(): val1 = StrOption('val1', "", 'val', properties=('disabled',)) val2 = StrOption('val2', "", Calculation(return_value, Params(value=ParamOption(val1)))) val3 = StrOption('val3', "", Calculation(return_value, Params(value=ParamOption(val1, True)))) maconfig = OptionDescription('rootconfig', '', [val1, val2, val3]) async with await Config(maconfig) as cfg: await cfg.property.read_only() with pytest.raises(ConfigError): await cfg.option('val2').value.get() await cfg.option('val3').value.get() is None assert not await list_sessions() @pytest.mark.asyncio async def test_callback_symlink(config_type): val1 = StrOption('val1', "", 'val') val2 = SymLinkOption('val2', val1) val3 = StrOption('val3', "", Calculation(return_value, Params(ParamOption(val2)))) maconfig = OptionDescription('rootconfig', '', [val1, val2, val3]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == 'val' assert await cfg.option('val2').value.get() == 'val' assert await cfg.option('val3').value.get() == 'val' await cfg.option('val1').value.set('new-val') assert await cfg.option('val1').value.get() == 'new-val' assert await cfg.option('val3').value.get() == 'new-val' await cfg.option('val1').value.reset() assert await cfg.option('val1').value.get() == 'val' assert await cfg.option('val3').value.get() == 'val' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_list(): val1 = StrOption('val1', "", Calculation(return_list)) maconfig = OptionDescription('rootconfig', '', [val1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(ValueError): await cfg.option('val1').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_list2(): val1 = StrOption('val1', "", Calculation(return_list)) val2 = StrOption('val2', "", Calculation(return_value, Params(ParamOption(val1)))) maconfig = OptionDescription('rootconfig', '', [val1, val2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(ValueError): await cfg.option('val1').value.get() #await cfg.val2 with pytest.raises(ValueError): await cfg.option('val2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_multi(config_type): val1 = StrOption('val1', "", [Calculation(return_val)], multi=True) maconfig = OptionDescription('rootconfig', '', [val1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == ['val'] await cfg.option('val1').value.set(['new-val']) assert await cfg.option('val1').value.get() == ['new-val'] await cfg.option('val1').value.set(['new-val', 'new-val2']) assert await cfg.option('val1').value.get() == ['new-val', 'new-val2'] await cfg.option('val1').value.reset() assert await cfg.option('val1').value.get() == ['val'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_multi_value(config_type): val1 = StrOption('val1', "", ['val'], multi=True) option = ParamOption(val1) params1 = Params((option,)) value = ParamValue('yes') params2 = Params((value,)) params3 = Params((option, value)) val2 = StrOption('val2', "", Calculation(return_value, params1), multi=True) val3 = StrOption('val3', "", [Calculation(return_value, params2)], multi=True) val4 = StrOption('val4', "", Calculation(return_list2, params3), multi=True) maconfig = OptionDescription('rootconfig', '', [val1, val2, val3, val4]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == ['val'] assert await cfg.option('val2').value.get() == ['val'] assert await cfg.option('val4').value.get() == ['val', 'yes'] await cfg.option('val1').value.set(['new-val']) assert await cfg.option('val1').value.get() == ['new-val'] assert await cfg.option('val2').value.get() == ['new-val'] assert await cfg.option('val4').value.get() == ['new-val', 'yes'] await cfg.option('val1').value.set(['new-val', 'new-val2']) assert await cfg.option('val1').value.get() == ['new-val', 'new-val2'] assert await cfg.option('val2').value.get() == ['new-val', 'new-val2'] assert await cfg.option('val4').value.get() == ['new-val', 'new-val2', 'yes'] await cfg.option('val1').value.reset() assert await cfg.option('val1').value.get() == ['val'] assert await cfg.option('val2').value.get() == ['val'] assert await cfg.option('val3').value.get() == ['yes'] assert await cfg.option('val4').value.get() == ['val', 'yes'] await cfg.option('val2').value.set(['val', 'new']) assert await cfg.option('val1').value.get() == ['val'] assert await cfg.option('val2').value.get() == ['val', 'new'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_multi_list(config_type): val1 = StrOption('val1', "", Calculation(return_list), multi=True, properties=('notunique',)) maconfig = OptionDescription('rootconfig', '', [val1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == ['val', 'val'] await cfg.option('val1').value.set(['new-val']) assert await cfg.option('val1').value.get() == ['new-val'] await cfg.option('val1').value.set(['new-val', 'new-val2']) assert await cfg.option('val1').value.get() == ['new-val', 'new-val2'] await cfg.option('val1').value.reset() assert await cfg.option('val1').value.get() == ['val', 'val'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_multi_list_extend(config_type): val1 = StrOption('val1', "", Calculation(return_list2, Params((ParamValue(['1', '2', '3']), ParamValue(['4', '5'])))), multi=True) maconfig = OptionDescription('rootconfig', '', [val1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1').value.get() == ['1', '2', '3', '4', '5'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_multi_callback(config_type): val1 = StrOption('val1', "", [Calculation(return_val)], multi=True) interface1 = OptionDescription('val1', '', [val1]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1.val1').value.get() == ['val'] await cfg.option('val1.val1').value.set(['val1', undefined]) assert await cfg.option('val1.val1').value.get() == ['val1', None] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_multi_callback_default(config_type): val1 = StrOption('val1', "", default_multi=Calculation(return_val), multi=True) interface1 = OptionDescription('val1', '', [val1]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1.val1').value.get() == [] await cfg.option('val1.val1').value.set(['val1', undefined]) assert await cfg.option('val1.val1').value.get() == ['val1', 'val'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader(config_type): val1 = StrOption('val1', "", default=[Calculation(return_val)], default_multi=Calculation(return_val), multi=True, properties=('notunique',)) val2 = StrOption('val2', "", multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1.val1').value.get() == ['val'] await cfg.option('val1.val1').value.set([undefined, undefined]) assert await cfg.option('val1.val1').value.get() == ['val', 'val'] assert await cfg.option('val1.val2', 0).value.get() == None assert await cfg.option('val1.val2', 1).value.get() == None assert not await list_sessions() @pytest.mark.asyncio async def test_callback_follower(config_type): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", Calculation(return_value3, Params(ParamValue(['string', 'new']), {'index': ParamIndex()})), multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) await cfg.option('val1.val1').value.set(['val']) assert await cfg.option('val1.val2', 0).value.get() == 'string' await cfg.option('val1.val1').value.set(['val', 'val1']) assert await cfg.option('val1.val2', 0).value.get() == 'string' assert await cfg.option('val1.val2', 1).value.get() == 'new' await cfg.option('val1.val1').value.set(['val', 'val1', 'val2']) assert await cfg.option('val1.val2', 0).value.get() == 'string' assert await cfg.option('val1.val2', 1).value.get() == 'new' assert await cfg.option('val1.val2', 2).value.get() == None await cfg.option('val1.val1').value.set(['val', 'val1', 'val2', 'val3']) assert await cfg.option('val1.val2', 0).value.get() == 'string' assert await cfg.option('val1.val2', 1).value.get() == 'new' assert await cfg.option('val1.val2', 2).value.get() == None assert await cfg.option('val1.val2', 3).value.get() == None assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader2(config_type): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", multi=True, default_multi='val2') val3 = StrOption('val3', "", Calculation(calc_value, Params(ParamOption(val2), {'index': ParamIndex()})), multi=True) val4 = StrOption('val4', "", Calculation(calc_value, Params(ParamOption(val3), {'index': ParamIndex()})), multi=True) interface1 = Leadership('val1', '', [val1, val2, val3, val4]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) await cfg.option('val1.val1').value.set(['val']) assert await cfg.option('val1.val4', 0).value.get() == 'val2' assert await cfg.option('val1.val3', 0).value.get() == 'val2' assert await cfg.option('val1.val2', 0).value.get() == 'val2' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader_mandatory1(config_type): val = StrOption('val', "", default='val') val1 = StrOption('val1', "", Calculation(return_value2, Params(ParamOption(val))), properties=('mandatory',), multi=True) val3 = StrOption('val3', "", Calculation(return_index, Params(ParamOption(val1), {'index': ParamIndex()})), properties=('mandatory',), multi=True) val4 = StrOption('val4', "", Calculation(return_index, Params(ParamOption(val1), {'index': ParamIndex()})), properties=('mandatory',), multi=True) interface1 = Leadership('val1', '', [val1, val3, val4]) maconfig = OptionDescription('rootconfig', '', [val, interface1]) async with await Config(maconfig) as cfg_ori: await cfg_ori.property.read_only() cfg = await get_config(cfg_ori, config_type) assert await cfg.option('val1.val3', 0).value.get() == 'val' assert await cfg.option('val1.val4', 0).value.get() == 'val' assert await cfg.option('val1.val1').value.get() == ['val'] if config_type == 'tiramisu-api': await cfg.send() await cfg_ori.property.read_write() cfg = await get_config(cfg_ori, config_type) await cfg.option('val1.val1').value.set([undefined, 'val3']) if config_type == 'tiramisu-api': await cfg.send() await cfg_ori.property.read_only() cfg = await get_config(cfg_ori, config_type) assert await cfg.option('val1.val1').value.get() == ['val', 'val3'] assert await cfg.option('val1.val3', 0).value.get() == 'val' if config_type != 'tiramisu-api': # FIXME with pytest.raises(PropertiesOptionError): await cfg.option('val1.val3', 1).value.get() with pytest.raises(PropertiesOptionError): await cfg.option('val1.val4', 1).value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader_mandatory2(config_type): val = StrOption('val', "", default='val') val_ = StrOption('val_', "", default='val_') val1 = StrOption('val1', "", Calculation(return_index, Params(ParamOption(val), {'val2': ParamOption(val_)})), properties=('mandatory',), multi=True) val3 = StrOption('val3', "", Calculation(return_index, Params(ParamOption(val1), {'val2': ParamOption(val_), 'index': ParamIndex()})), properties=('mandatory',), multi=True) val4 = StrOption('val4', "", Calculation(return_index, Params(ParamOption(val1), {'val2': ParamOption(val_), 'index': ParamIndex()})), properties=('mandatory',), multi=True) interface1 = Leadership('val1', '', [val1, val3, val4]) maconfig = OptionDescription('rootconfig', '', [val, val_, interface1]) async with await Config(maconfig) as cfg_ori: await cfg_ori.property.read_only() cfg = await get_config(cfg_ori, config_type) assert await cfg.option('val1.val3', 0).value.get() == 'val' assert await cfg.option('val1.val3', 1).value.get() == 'val_' assert await cfg.option('val1.val4', 0).value.get() == 'val' assert await cfg.option('val1.val4', 1).value.get() == 'val_' assert await cfg.option('val1.val1').value.get() == ['val', 'val_'] await cfg_ori.property.read_write() cfg = await get_config(cfg_ori, config_type) await cfg.option('val1.val1').value.set(['val', 'val_', 'val3']) assert await cfg.option('val1.val1').value.get() == ['val', 'val_', 'val3'] await cfg_ori.property.read_only() cfg = await get_config(cfg_ori, config_type) assert await cfg.option('val1.val3', 0).value.get() == 'val' assert await cfg.option('val1.val3', 1).value.get() == 'val_' assert await cfg.option('val1.val4', 0).value.get() == 'val' assert await cfg.option('val1.val4', 1).value.get() == 'val_' if config_type != 'tiramisu-api': # FIXME with pytest.raises(PropertiesOptionError): await cfg.option('val1.val3', 2).value.get() with pytest.raises(PropertiesOptionError): await cfg.option('val1.val4', 2).value.get() assert await cfg.option('val1.val1').value.get() == ['val', 'val_', 'val3'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader_mandatory3(config_type): val = StrOption('val', "", default='val') val_ = StrOption('val_', "", default='val_') val1 = StrOption('val1', "", Calculation(return_value2, Params(ParamOption(val), {'val': ParamOption(val_)})), properties=('mandatory',), multi=True) val3 = StrOption('val3', "", Calculation(calc_value, Params(ParamOption(val1), {'index': ParamIndex()})), properties=('mandatory',), multi=True) val4 = StrOption('val4', "", Calculation(calc_value, Params(ParamOption(val1), {'index': ParamIndex()})), properties=('mandatory',), multi=True) interface1 = Leadership('val1', '', [val1, val3, val4]) maconfig = OptionDescription('rootconfig', '', [val, val_, interface1]) async with await Config(maconfig) as cfg_ori: await cfg_ori.property.read_only() cfg = await get_config(cfg_ori, config_type) assert await cfg.option('val1.val3', 0).value.get() == 'val' assert await cfg.option('val1.val3', 1).value.get() == 'val_' assert await cfg.option('val1.val4', 0).value.get() == 'val' assert await cfg.option('val1.val4', 1).value.get() == 'val_' assert await cfg.option('val1.val1').value.get() == ['val', 'val_'] await cfg_ori.property.read_write() cfg = await get_config(cfg_ori, config_type) await cfg.option('val1.val1').value.set(['val', 'val_', 'val3']) await cfg_ori.property.read_only() cfg = await get_config(cfg_ori, config_type) assert await cfg.option('val1.val3', 0).value.get() == 'val' assert await cfg.option('val1.val3', 1).value.get() == 'val_' assert await cfg.option('val1.val3', 2).value.get() == 'val3' assert await cfg.option('val1.val4', 0).value.get() == 'val' assert await cfg.option('val1.val4', 1).value.get() == 'val_' assert await cfg.option('val1.val4', 2).value.get() == 'val3' assert await cfg.option('val1.val1').value.get() == ['val', 'val_', 'val3'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader_mandatory4(config_type): val = StrOption('val', "", default='val') val1 = StrOption('val1', "", Calculation(return_value2, Params(ParamOption(val))), properties=('mandatory',), multi=True) val3 = StrOption('val3', "", Calculation(calc_value, Params(ParamOption(val1), {'index': ParamIndex()})), properties=('mandatory',), multi=True) val4 = StrOption('val4', "", Calculation(calc_value, Params(ParamOption(val1), {'index': ParamIndex()})), properties=('mandatory',), multi=True) interface1 = Leadership('val1', '', [val1, val3, val4]) maconfig = OptionDescription('rootconfig', '', [val, interface1]) async with await Config(maconfig) as cfg_ori: await cfg_ori.property.read_only() cfg = await get_config(cfg_ori, config_type) #raises(IndexError, "await cfg.option('val1.val3').value.get()") assert await cfg.option('val1.val3', 0).value.get() == 'val' assert await cfg.option('val1.val4', 0).value.get() == 'val' assert await cfg.option('val1.val1').value.get() == ['val'] await cfg_ori.property.read_write() cfg = await get_config(cfg_ori, config_type) await cfg.option('val1.val1').value.set(['val', 'val3']) await cfg_ori.property.read_only() cfg = await get_config(cfg_ori, config_type) assert await cfg.option('val1.val1').value.get() == ['val', 'val3'] assert await cfg.option('val1.val3', 0).value.get() == 'val' assert await cfg.option('val1.val3', 1).value.get() == 'val3' assert await cfg.option('val1.val4', 0).value.get() == 'val' assert await cfg.option('val1.val4', 1).value.get() == 'val3' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader3(): val1 = StrOption('val1', "", multi=True, properties=('mandatory', 'empty')) val2 = StrOption('val2', "", multi=True, default_multi='val2', properties=('expert',)) val3 = StrOption('val3', "", Calculation(calc_value, Params(ParamOption(val2), {'index': ParamIndex()})), multi=True) val4 = StrOption('val4', "", Calculation(calc_value, Params(ParamOption(val3), {'index': ParamIndex()})), multi=True) interface1 = Leadership('val1', '', [val1, val2, val3, val4]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() # FIXME cfg = await get_config(cfg, config_type) assert list(await cfg.value.mandatory()) == ['val1.val1'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader4(): val1 = StrOption('val1', "", ['val1'], multi=True, properties=('mandatory',)) val2 = StrOption('val2', "", multi=True, default_multi='val2', properties=('expert', 'mandatory')) val3 = StrOption('val3', "", Calculation(calc_value, Params(ParamOption(val2), {'index': ParamIndex()})), multi=True) val4 = StrOption('val4', "", Calculation(calc_value, Params(ParamOption(val3), {'index': ParamIndex()})), multi=True) interface1 = Leadership('val1', '', [val1, val2, val3, val4]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() # FIXME cfg = await get_config(cfg, config_type) await cfg.property.add('expert') await cfg.permissive.add('expert') assert list(await cfg.value.mandatory()) == [] assert not await list_sessions() @pytest.mark.asyncio async def test_consistency_leader_and_followers_leader_mandatory_transitive(): #default value val1 = IPOption('val1', "", ['192.168.0.1'], multi=True, properties=('mandatory',)) val2 = NetmaskOption('val2', "", multi=True, default_multi='255.255.255.0', properties=('disabled', 'mandatory'), validators=[Calculation(valid_ip_netmask, Params((ParamOption(val1), ParamSelfOption())))]) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() # FIXME cfg = await get_config(cfg, config_type) try: await cfg.option('val1.val2', 0).value.get() except PropertiesOptionError as error: assert str(error) == str(_('cannot access to {0} "{1}" because has {2} {3}').format('option', 'val2', _('property'), '"disabled"')) else: raise Exception('must raises') assert list(await cfg.value.mandatory()) == [] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader_list(config_type): val1 = StrOption('val1', "", Calculation(return_list), multi=True, properties=('notunique',)) val2 = StrOption('val2', "", multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1.val1').value.get() == ['val', 'val'] assert await cfg.option('val1.val2', 0).value.get() == None assert await cfg.option('val1.val2', 1).value.get() == None await cfg.option('val1.val1').value.set(['val', 'val', undefined]) assert await cfg.option('val1.val1').value.get() == ['val', 'val', None] assert await cfg.option('val1.val2', 0).value.get() == None assert await cfg.option('val1.val2', 1).value.get() == None assert await cfg.option('val1.val2', 2).value.get() == None await cfg.option('val1.val1').value.reset() assert await cfg.option('val1.val1').value.get() == ['val', 'val'] assert await cfg.option('val1.val2', 0).value.get() == None assert await cfg.option('val1.val2', 1).value.get() == None await cfg.option('val1.val1').value.pop(1) assert await cfg.option('val1.val1').value.get() == ['val'] assert await cfg.option('val1.val2', 0).value.get() == None assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader_follower_list(config_type): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", Calculation(return_list), multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1.val1').value.get() == [] if config_type == 'tiramisu-api': # when "tiramisu-api", raise when set and not in get function with pytest.raises(ConfigError): await cfg.option('val1.val1').value.set(['val1']) else: await cfg.option('val1.val1').value.set(['val1']) with pytest.raises(LeadershipError): await cfg.option('val1.val2', 0).value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_follower(config_type): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", Calculation(return_val), multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1.val1').value.get() == [] # await cfg.option('val1.val1').value.set(['val1']) assert await cfg.option('val1.val1').value.get() == ['val1'] assert await cfg.option('val1.val2', 0).value.get() == 'val' # await cfg.option('val1.val1').value.set(['val1', 'val2']) assert await cfg.option('val1.val1').value.get() == ['val1', 'val2'] assert await cfg.option('val1.val2', 0).value.get() == 'val' assert await cfg.option('val1.val2', 1).value.get() == 'val' # await cfg.option('val1.val1').value.set(['val1', 'val2', 'val3']) assert await cfg.option('val1.val1').value.get() == ['val1', 'val2', 'val3'] assert await cfg.option('val1.val2', 0).value.get() == 'val' assert await cfg.option('val1.val2', 1).value.get() == 'val' assert await cfg.option('val1.val2', 2).value.get() == 'val' # await cfg.option('val1.val1').value.pop(2) assert await cfg.option('val1.val1').value.get() == ['val1', 'val2'] assert await cfg.option('val1.val2', 0).value.get() == 'val' assert await cfg.option('val1.val2', 1).value.get() == 'val' # await cfg.option('val1.val2', 0).value.set('val2') await cfg.option('val1.val2', 1).value.set('val2') assert await cfg.option('val1.val2', 0).value.get() == 'val2' assert await cfg.option('val1.val2', 1).value.get() == 'val2' # await cfg.option('val1.val1').value.set(['val1', 'val2', 'val3']) assert await cfg.option('val1.val2', 0).value.get() == 'val2' assert await cfg.option('val1.val2', 1).value.get() == 'val2' assert await cfg.option('val1.val2', 2).value.get() == 'val' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers(): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", Calculation(return_val), multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_follower_cal(config_type): val3 = StrOption('val3', "", multi=True) val1 = StrOption('val1', "", Calculation(return_value, Params(ParamOption(val3))), multi=True) val2 = StrOption('val2', "", Calculation(return_val), multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1, val3]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) # assert await cfg.option('val3').value.get() == [] assert await cfg.option('val1.val1').value.get() == [] # await cfg.option('val1.val1').value.set(['val1']) await cfg.option('val3').value.set(['val1']) assert await cfg.option('val1.val1').value.get() == ['val1'] assert await cfg.option('val1.val2', 0).value.get() == 'val' # await cfg.option('val1.val1').value.reset() await cfg.option('val1.val2', 0).value.set('val') # await cfg.option('val3').value.set(['val1', 'val2']) assert await cfg.option('val1.val2', 0).value.get() == 'val' assert await cfg.option('val1.val2', 1).value.get() == 'val' assert await cfg.option('val1.val1').value.get() == ['val1', 'val2'] # len of follower is higher than leader's one await cfg.option('val1.val2', 0).value.set('val1') await cfg.option('val1.val2', 1).value.set('val2') if config_type == 'tiramisu-api': # when "tiramisu-api", raise when set and not in get function with pytest.raises(ConfigError): await cfg.option('val3').value.set(['val1']) else: await cfg.option('val3').value.set(['val1']) assert await cfg.option('val1.val1').value.get() == ['val1'] with pytest.raises(LeadershipError): await cfg.option('val1.val2', 0).value.get() # await cfg.option('val3').value.set(['val1', 'val2', 'val3']) assert await cfg.option('val1.val2', 0).value.get() == 'val1' assert await cfg.option('val1.val2', 1).value.get() == 'val2' assert await cfg.option('val1.val2', 2).value.get() == 'val' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader_disabled(): #properties must be transitive val1 = StrOption('val1', "", ['val1'], multi=True) val2 = StrOption('val2', "", multi=True) interface1 = Leadership('val1', '', [val1, val2], properties=('disabled',)) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(PropertiesOptionError): await cfg.option('val1.val1').value.get() with pytest.raises(PropertiesOptionError): await cfg.option('val1.val1').value.set(['yes']) with pytest.raises(PropertiesOptionError): await cfg.option('val1.val2', 0).value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_leader_callback_disabled(): val0 = StrOption('val0', "", multi=True, properties=('disabled',)) val1 = StrOption('val1', "", Calculation(return_value, Params(ParamOption(val0))), multi=True) val2 = StrOption('val2', "", multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1, val0]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(ConfigError): await cfg.option('val1.val1').value.get() with pytest.raises(ConfigError): await cfg.option('val1.val2').value.get() await cfg.property.pop('disabled') await cfg.option('val1.val1').value.set([]) await cfg.property.add('disabled') assert await cfg.option('val1.val1').value.get() == [] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_follower_disabled(): val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", multi=True, properties=('disabled',)) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1]) async with await Config(maconfig) as cfg: await cfg.property.read_write() assert await cfg.option('val1.val1').value.get() == [] await cfg.option('val1.val1').value.set(['yes']) assert await cfg.option('val1.val1').value.get() == ['yes'] await cfg.property.pop('disabled') assert await cfg.option('val1.val2', 0).value.get() == None await cfg.option('val1.val2', 0).value.set('no') await cfg.option('val1.val1').value.set(['yes', 'yes2', 'yes3']) await cfg.option('val1.val2', 2).value.set('no1') assert await cfg.option('val1.val2', 0).value.get() == 'no' assert await cfg.option('val1.val2', 1).value.get() == None assert await cfg.option('val1.val2', 2).value.get() == 'no1' await cfg.property.add('disabled') await cfg.option('val1.val1').value.pop(0) assert await cfg.option('val1.val1').value.get() == ['yes2', 'yes3'] await cfg.property.pop('disabled') assert await cfg.option('val1.val2', 0).value.get() == None assert await cfg.option('val1.val2', 1).value.get() == 'no1' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_follower_callback_disabled(): val0 = StrOption('val0', "", multi=True, properties=('disabled',)) val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", Calculation(return_value, Params(ParamOption(val0))), multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1, val0]) async with await Config(maconfig) as cfg: await cfg.property.read_write() assert await cfg.option('val1.val1').value.get() == [] await cfg.option('val1.val1').value.set(['yes']) assert await cfg.option('val1.val1').value.get() == ['yes'] await cfg.property.pop('disabled') await cfg.option('val1.val2', 0).value.set('no') await cfg.option('val1.val1').value.set(['yes', 'yes1']) assert await cfg.option('val1.val2', 0).value.get() == 'no' await cfg.property.add('disabled') await cfg.option('val1.val1').value.pop(1) assert not await list_sessions() @pytest.mark.asyncio async def test_callback_leader_and_followers_value(): val4 = StrOption('val4', '', multi=True, default=['val10', 'val11']) val1 = StrOption('val1', "", multi=True) val2 = StrOption('val2', "", Calculation(return_value, Params(ParamOption(val1))), multi=True) val3 = StrOption('val3', "", Calculation(return_value, Params(ParamValue('yes'))), multi=True) val5 = StrOption('val5', "", Calculation(return_value, Params(ParamOption(val4))), multi=True) val6 = StrOption('val6', "", Calculation(return_value, Params(ParamOption(val5))), multi=True) interface1 = Leadership('val1', '', [val1, val2, val3, val5, val6]) maconfig = OptionDescription('rootconfig', '', [interface1, val4]) async with await Config(maconfig) as cfg: await cfg.property.read_write() await cfg.option('val4').value.get() == ['val10', 'val11'] assert await cfg.option('val1.val1').value.get() == [] #with pytest.raises(LeadershipError): # await cfg.val1.val1") #with pytest.raises(LeadershipError): # await cfg.val1.val2") #with pytest.raises(LeadershipError): # await cfg.val1.val3") #with pytest.raises(LeadershipError): # await cfg.val1.val5") #with pytest.raises(LeadershipError): # await cfg.val1.val6") # #default calculation has greater length #with pytest.raises(LeadershipError): # await cfg.option('val1.val1').value.set(['val1']") # await cfg.option('val1.val1').value.set(['val1', 'val2']) assert await cfg.option('val1.val1').value.get() == ['val1', 'val2'] assert await cfg.option('val1.val2', 0).value.get() == 'val1' assert await cfg.option('val1.val2', 1).value.get() == 'val2' assert await cfg.option('val1.val3', 0).value.get() == 'yes' assert await cfg.option('val1.val3', 1).value.get() == 'yes' with pytest.raises(LeadershipError): await cfg.option('val1.val5', 0).value.get() with pytest.raises(LeadershipError): await cfg.option('val1.val5', 1).value.get() with pytest.raises(LeadershipError): await cfg.option('val1.val6', 0).value.get() with pytest.raises(LeadershipError): await cfg.option('val1.val6', 1).value.get() # await cfg.option('val1.val1').value.set(['val1', 'val2', 'val3']) assert await cfg.option('val1.val1').value.get() == ['val1', 'val2', 'val3'] assert await cfg.option('val1.val2', 0).value.get() == 'val1' assert await cfg.option('val1.val2', 1).value.get() == 'val2' assert await cfg.option('val1.val2', 2).value.get() == 'val3' assert await cfg.option('val1.val3', 0).value.get() == 'yes' assert await cfg.option('val1.val3', 1).value.get() == 'yes' assert await cfg.option('val1.val3', 2).value.get() == 'yes' with pytest.raises(LeadershipError): await cfg.option('val1.val5', 2).value.get() with pytest.raises(LeadershipError): await cfg.option('val1.val6', 2).value.get() # await cfg.option('val1.val1').value.pop(2) assert await cfg.option('val1.val1').value.get() == ['val1', 'val2'] assert await cfg.option('val1.val2', 0).value.get() == 'val1' assert await cfg.option('val1.val2', 1).value.get() == 'val2' assert await cfg.option('val1.val3', 0).value.get() == 'yes' assert await cfg.option('val1.val3', 1).value.get() == 'yes' # await cfg.option('val1.val2', 0).value.set('val2') await cfg.option('val1.val2', 1).value.set('val2') await cfg.option('val1.val3', 0).value.set('val2') await cfg.option('val1.val3', 1).value.set('val2') await cfg.option('val1.val5', 0).value.set('val2') await cfg.option('val1.val5', 1).value.set('val2') assert await cfg.option('val1.val2', 0).value.get() == 'val2' assert await cfg.option('val1.val2', 1).value.get() == 'val2' assert await cfg.option('val1.val3', 0).value.get() == 'val2' assert await cfg.option('val1.val3', 1).value.get() == 'val2' assert await cfg.option('val1.val5', 0).value.get() == 'val2' assert await cfg.option('val1.val5', 1).value.get() == 'val2' assert await cfg.option('val1.val6', 0).value.get() == 'val2' assert await cfg.option('val1.val6', 1).value.get() == 'val2' # await cfg.option('val1.val1').value.set(['val1', 'val2', 'val3']) assert await cfg.option('val1.val2', 2).value.get() == 'val3' assert await cfg.option('val1.val3', 2).value.get() == 'yes' assert not await list_sessions() @pytest.mark.asyncio async def test_callback_different_type(config_type): val = IntOption('val', "", default=2) val_ = IntOption('val_', "", default=3) val1 = IntOption('val1', "", multi=True) val2 = IntOption('val2', "", Calculation(return_calc, Params((ParamOption(val), ParamOption(val1)), {'k': ParamOption(val_)})), multi=True) interface1 = Leadership('val1', '', [val1, val2]) maconfig = OptionDescription('rootconfig', '', [interface1, val, val_]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val1.val1').value.get() == [] await cfg.option('val1.val1').value.set([1]) assert await cfg.option('val1.val1').value.get() == [1] assert await cfg.option('val1.val2', 0).value.get() == 6 await cfg.option('val1.val1').value.set([1, 3]) assert await cfg.option('val1.val1').value.get() == [1, 3] assert await cfg.option('val1.val2', 0).value.get() == 6 assert await cfg.option('val1.val2', 1).value.get() == 8 await cfg.option('val1.val1').value.set([1, 3, 5]) assert await cfg.option('val1.val1').value.get() == [1, 3, 5] assert await cfg.option('val1.val2', 0).value.get() == 6 assert await cfg.option('val1.val2', 1).value.get() == 8 assert await cfg.option('val1.val2', 2).value.get() == 10 assert not await list_sessions() @pytest.mark.asyncio async def test_callback_hidden(): opt1 = BoolOption('opt1', '') opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1)))) od1 = OptionDescription('od1', '', [opt1], properties=('hidden',)) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(PropertiesOptionError): await cfg.option('od1.opt1').value.get() # do not raise, forcepermissive await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_hidden_permissive(): opt1 = BoolOption('opt1', '') opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1)))) od1 = OptionDescription('od1', '', [opt1], properties=('hidden',)) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.permissive.add('hidden') await cfg.property.read_write() with pytest.raises(PropertiesOptionError): await cfg.option('od1.opt1').value.get() await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_hidden_permissive_callback(): opt1 = BoolOption('opt1', '') opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1, True)))) od1 = OptionDescription('od1', '', [opt1], properties=('hidden',)) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(PropertiesOptionError): await cfg.option('od1.opt1').value.get() await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_two_disabled(): opt1 = BoolOption('opt1', '', properties=('disabled',)) opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1))), properties=('disabled',)) od1 = OptionDescription('od1', '', [opt1]) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(PropertiesOptionError): await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_two_disabled2(): opt1 = BoolOption('opt1', '', properties=('hidden',)) opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1))), properties=('hidden',)) od1 = OptionDescription('od1', '', [opt1]) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() await cfg.permissive.add('hidden') with pytest.raises(PropertiesOptionError): await cfg.option('od2.opt2').value.get() assert await cfg.forcepermissive.option('od2.opt2').owner.isdefault() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_calculating_invalid(): opt1 = IntOption('opt1', '', 1) opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1)))) od1 = OptionDescription('od1', '', [opt1]) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(ValueError): await cfg.option('od2.opt2').value.get() await cfg.unrestraint.option('od2.opt2').property.add('disabled') with pytest.raises(PropertiesOptionError): await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_calculating_disabled(): opt1 = BoolOption('opt1', '', properties=('disabled',)) opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1)))) od1 = OptionDescription('od1', '', [opt1]) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(ConfigError): await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_calculating_mandatory(): opt1 = BoolOption('opt1', '', properties=('disabled',)) opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1))), properties=('mandatory',)) od1 = OptionDescription('od1', '', [opt1]) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_only() with pytest.raises(ConfigError): await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_calculating_mandatory_multi(): opt1 = BoolOption('opt1', '', multi=True, properties=('disabled',)) opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1))), properties=('mandatory',), multi=True) od1 = OptionDescription('od1', '', [opt1]) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_only() with pytest.raises(ConfigError): await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_two_disabled_multi(): opt1 = BoolOption('opt1', '', properties=('disabled',)) opt2 = BoolOption('opt2', '', Calculation(return_value, Params(ParamOption(opt1))), properties=('disabled',), multi=True) od1 = OptionDescription('od1', '', [opt1]) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() with pytest.raises(PropertiesOptionError): await cfg.option('od2.opt2').value.get() assert not await list_sessions() @pytest.mark.asyncio async def test_callback_multi_list_params(config_type): val1 = StrOption('val1', "", multi=True, default=['val1', 'val2']) val2 = StrOption('val2', "", Calculation(return_list, Params(ParamOption(val1))), multi=True, properties=('notunique',)) oval2 = OptionDescription('val2', '', [val2]) maconfig = OptionDescription('rootconfig', '', [val1, oval2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val2.val2').value.get() == ['val', 'val'] assert not await list_sessions() @pytest.mark.asyncio async def test_callback_multi_list_params_key(config_type): val1 = StrOption('val1', "", multi=True, default=['val1', 'val2']) val2 = StrOption('val2', "", Calculation(return_list, Params(kwargs={'value': ParamOption(val1)})), multi=True, properties=('notunique',)) oval2 = OptionDescription('val2', '', [val2]) maconfig = OptionDescription('rootconfig', '', [val1, oval2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.option('val2.val2').value.get() == ['val', 'val'] assert not await list_sessions() @pytest.mark.asyncio async def test_leadership_callback_description(config_type): st1 = StrOption('st1', "", multi=True) st2 = StrOption('st2', "", Calculation(return_value, Params(ParamOption(st1))), multi=True) stm = Leadership('st1', '', [st1, st2]) st = OptionDescription('st', '', [stm]) od = OptionDescription('od', '', [st]) od2 = OptionDescription('od', '', [od]) async with await Config(od2) as cfg: cfg = await get_config(cfg, config_type) owner = await cfg.owner.get() assert await cfg.option('od.st.st1.st1').value.get() == [] assert await cfg.option('od.st.st1.st1').owner.isdefault() ## await cfg.option('od.st.st1.st1').value.set(['yes']) await cfg.option('od.st.st1.st2', 0).value.set('yes') assert await cfg.option('od.st.st1.st1').owner.get() == owner assert await cfg.option('od.st.st1.st2', 0).owner.get() == owner assert not await list_sessions() @pytest.mark.asyncio async def test_callback_raise(): opt1 = BoolOption('opt1', 'Option 1', Calculation(return_raise)) opt2 = BoolOption('opt2', 'Option 2', Calculation(return_valueerror)) od1 = OptionDescription('od1', '', [opt1]) od2 = OptionDescription('od2', '', [opt2]) maconfig = OptionDescription('rootconfig', '', [od1, od2]) async with await Config(maconfig) as cfg: await cfg.property.read_write() try: await cfg.option('od1.opt1').value.get() except ConfigError as err: assert '"Option 1"' in str(err) try: await cfg.option('od2.opt2').value.get() except ConfigError as err: assert '"Option 2"' in str(err) assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_simple(config_type): val1 = StrOption('val1', '', 'val1') val2 = StrOption('val2', '', Calculation(calc_value, Params(ParamOption(val1)))) od = OptionDescription('root', '', [val1, val2]) async with await Config(od) as cfg: cfg = await get_config(cfg, config_type) assert await cfg.value.dict() == {'val1': 'val1', 'val2': 'val1'} assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_multi(config_type): val1 = StrOption('val1', "", 'val1') val2 = StrOption('val2', "", 'val2') val3 = StrOption('val3', "", Calculation(calc_value, Params((ParamOption(val1), ParamOption(val2)), multi=ParamValue(True))), multi=True) od = OptionDescription('root', '', [val1, val2, val3]) async with await Config(od) as cfg: cfg = await get_config(cfg, config_type) assert await cfg.value.dict() == {'val1': 'val1', 'val2': 'val2', 'val3': ['val1', 'val2']} assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_disabled(): val1 = StrOption('val1', '', 'val1') val2 = StrOption('val2', '', Calculation(calc_value, Params(ParamOption(val1, True), default=ParamValue('default_value')))) od = OptionDescription('root', '', [val1, val2]) async with await Config(od) as cfg: await cfg.property.read_write() assert await cfg.value.dict() == {'val1': 'val1', 'val2': 'val1'} await cfg.option('val1').property.add('disabled') assert await cfg.value.dict() == {'val2': 'default_value'} assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_condition(config_type): boolean = BoolOption('boolean', '', True) val1 = StrOption('val1', '', 'val1') val2 = StrOption('val2', '', Calculation(calc_value, Params(ParamOption(val1, True), default=ParamValue('default_value'), condition=ParamOption(boolean), expected=ParamValue(True)))) od = OptionDescription('root', '', [boolean, val1, val2]) async with await Config(od) as cfg: await cfg.property.read_write() cfg = await get_config(cfg, config_type) assert await cfg.value.dict() == {'boolean': True, 'val1': 'val1', 'val2': 'val1'} await cfg.option('boolean').value.set(False) assert await cfg.value.dict() == {'boolean': False, 'val1': 'val1', 'val2': 'default_value'} assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_allow_none(config_type): val1 = StrOption('val1', "", 'val1') val2 = StrOption('val2', "") val3 = StrOption('val3', "", Calculation(calc_value, Params((ParamOption(val1), ParamOption(val2)), multi=ParamValue(True), allow_none=ParamValue(True))), multi=True) od = OptionDescription('root', '', [val1, val2, val3]) async with await Config(od) as cfg: cfg = await get_config(cfg, config_type) assert await cfg.value.dict() == {'val1': 'val1', 'val2': None, 'val3': ['val1', None]} assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_remove_duplicate(config_type): val1 = StrOption('val1', "", 'val1') val2 = StrOption('val2', "", 'val1') val3 = StrOption('val3', "", Calculation(calc_value, Params((ParamOption(val1), ParamOption(val2)), multi=ParamValue(True), remove_duplicate_value=ParamValue(True))), multi=True) od = OptionDescription('root', '', [val1, val2, val3]) async with await Config(od) as cfg: cfg = await get_config(cfg, config_type) assert await cfg.value.dict() == {'val1': 'val1', 'val2': 'val1', 'val3': ['val1']} assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_join(config_type): val1 = StrOption('val1', "", 'val1') val2 = StrOption('val2', "", 'val2') val3 = StrOption('val3', "") val4 = StrOption('val4', "", Calculation(calc_value, Params((ParamOption(val1), ParamOption(val2), ParamOption(val3)), join=ParamValue('.')))) od = OptionDescription('root', '', [val1, val2, val3, val4]) async with await Config(od) as cfg: cfg = await get_config(cfg, config_type) assert await cfg.value.dict() == {'val1': 'val1', 'val2': 'val2', 'val3': None, 'val4': None} await cfg.option('val3').value.set('val3') assert await cfg.value.dict() == {'val1': 'val1', 'val2': 'val2', 'val3': 'val3', 'val4': 'val1.val2.val3'} assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_min(): val1 = StrOption('val1', "", 'val1') val2 = StrOption('val2', "", 'val2') val3 = StrOption('val3', "", 'val3') val4 = StrOption('val4', "", Calculation(calc_value, Params((ParamOption(val1), ParamOption(val2), ParamOption(val3, True)), join=ParamValue('.'), min_args_len=ParamValue(3)))) od = OptionDescription('root', '', [val1, val2, val3, val4]) async with await Config(od) as cfg: await cfg.property.read_write() assert await cfg.value.dict() == {'val1': 'val1', 'val2': 'val2', 'val3': 'val3', 'val4': 'val1.val2.val3'} await cfg.option('val3').property.add('disabled') assert await cfg.value.dict() == {'val1': 'val1', 'val2': 'val2', 'val4': ''} assert not await list_sessions() @pytest.mark.asyncio async def test_calc_value_add(config_type): val1 = IntOption('val1', "", 1) val2 = IntOption('val2', "", 2) val3 = IntOption('val3', "", Calculation(calc_value, Params((ParamOption(val1), ParamOption(val2)), operator=ParamValue('add')))) od = OptionDescription('root', '', [val1, val2, val3]) async with await Config(od) as cfg: cfg = await get_config(cfg, config_type) assert await cfg.value.dict() == {'val1': 1, 'val2': 2, 'val3': 3} assert not await list_sessions()