247 lines
10 KiB
Python
247 lines
10 KiB
Python
# coding: utf-8
|
|
"frozen and hidden values"
|
|
from .autopath import do_autopath
|
|
do_autopath()
|
|
|
|
import pytest
|
|
|
|
from tiramisu.setting import owners, groups
|
|
from tiramisu import ChoiceOption, BoolOption, IntOption, FloatOption, \
|
|
StrOption, OptionDescription, SymLinkOption, Leadership, Config, \
|
|
Calculation, Params, ParamOption, ParamValue, calc_value, delete_session
|
|
from tiramisu.error import PropertiesOptionError, ConfigError
|
|
from tiramisu.storage import list_sessions
|
|
from .config import event_loop
|
|
|
|
|
|
def compare(calculated, expected):
|
|
def convert_list(val):
|
|
if isinstance(val, list):
|
|
val = tuple(val)
|
|
return val
|
|
# convert to tuple
|
|
for idx in range(len(calculated[0])):
|
|
right_idx = expected[0].index(calculated[0][idx])
|
|
for typ in range(4):
|
|
assert convert_list(calculated[typ][idx]) == expected[typ][right_idx]
|
|
|
|
|
|
#____________________________________________________________
|
|
#freeze
|
|
def make_description_freeze():
|
|
gcoption = ChoiceOption('name', 'GC name', ('ref', 'framework'), 'ref')
|
|
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], multi=True)
|
|
hidden_property = Calculation(calc_value,
|
|
Params(ParamValue('hidden'),
|
|
kwargs={'condition': ParamOption(booloption, raisepropertyerror=True),
|
|
'expected': ParamValue(True),
|
|
'default': ParamValue(None)}))
|
|
wantref_option = BoolOption('wantref', 'Test requires', default=False, properties=('force_store_value', hidden_property))
|
|
wantref2_option = BoolOption('wantref2', 'Test requires', default=False, properties=('force_store_value', 'hidden'))
|
|
wantref3_option = BoolOption('wantref3', 'Test requires', default=[False], multi=True, properties=('force_store_value',))
|
|
st2 = SymLinkOption('st2', wantref3_option)
|
|
hidden_property = Calculation(calc_value,
|
|
Params(ParamValue('hidden'),
|
|
kwargs={'condition': ParamOption(booloption, raisepropertyerror=True),
|
|
'expected': ParamValue(True),
|
|
'default': ParamValue(None)}))
|
|
wantframework_option = BoolOption('wantframework', 'Test requires',
|
|
default=False,
|
|
properties=(hidden_property,))
|
|
|
|
gcgroup = OptionDescription('gc', '', [gcoption, gcdummy, floatoption])
|
|
descr = OptionDescription('tiramisu', '', [gcgroup, booloption, objspaceoption,
|
|
wantref_option, wantref2_option, wantref3_option, st2, stroption,
|
|
wantframework_option,
|
|
intoption, boolop])
|
|
return descr
|
|
|
|
|
|
def return_val():
|
|
return 1
|
|
|
|
|
|
def return_val2(value):
|
|
return value
|
|
|
|
|
|
def return_val3(context, value):
|
|
return value
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_freeze_whole_config():
|
|
descr = make_description_freeze()
|
|
async with await Config(descr) as cfg:
|
|
await cfg.property.read_write()
|
|
await cfg.property.add('everything_frozen')
|
|
assert await cfg.option('gc.dummy').value.get() is False
|
|
prop = []
|
|
try:
|
|
await cfg.option('gc.dummy').value.set(True)
|
|
except PropertiesOptionError as err:
|
|
prop = err.proptype
|
|
assert 'frozen' in prop
|
|
assert await cfg.option('gc.dummy').value.get() is False
|
|
#
|
|
await cfg.property.pop('everything_frozen')
|
|
await cfg.option('gc.dummy').value.set(True)
|
|
assert await cfg.option('gc.dummy').value.get() is True
|
|
#
|
|
await cfg.property.add('everything_frozen')
|
|
owners.addowner("everythingfrozen2")
|
|
prop = []
|
|
try:
|
|
await cfg.option('gc.dummy').owner.set('everythingfrozen2')
|
|
except PropertiesOptionError as err:
|
|
prop = err.proptype
|
|
assert 'frozen' in prop
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_freeze_one_option():
|
|
"freeze an option "
|
|
descr = make_description_freeze()
|
|
async with await Config(descr) as cfg:
|
|
await cfg.property.read_write()
|
|
#freeze only one option
|
|
await cfg.option('gc.dummy').property.add('frozen')
|
|
assert await cfg.option('gc.dummy').value.get() is False
|
|
prop = []
|
|
try:
|
|
await cfg.option('gc.dummy').value.set(True)
|
|
except PropertiesOptionError as err:
|
|
prop = err.proptype
|
|
assert 'frozen' in prop
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_frozen_value():
|
|
"setattr a frozen value at the config level"
|
|
s = StrOption("string", "", default="string")
|
|
descr = OptionDescription("options", "", [s])
|
|
async with await Config(descr) as cfg:
|
|
await cfg.property.read_write()
|
|
await cfg.property.add('frozen')
|
|
await cfg.option('string').property.add('frozen')
|
|
prop = []
|
|
try:
|
|
await cfg.option('string').value.set('egg')
|
|
except PropertiesOptionError as err:
|
|
prop = err.proptype
|
|
assert 'frozen' in prop
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_freeze():
|
|
"freeze a whole configuration object"
|
|
descr = make_description_freeze()
|
|
async with await Config(descr) as cfg:
|
|
await cfg.property.read_write()
|
|
await cfg.property.add('frozen')
|
|
await cfg.option('gc.name').property.add('frozen')
|
|
prop = []
|
|
try:
|
|
await cfg.option('gc.name').value.set('framework')
|
|
except PropertiesOptionError as err:
|
|
prop = err.proptype
|
|
assert 'frozen' in prop
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_freeze_multi():
|
|
descr = make_description_freeze()
|
|
async with await Config(descr) as cfg:
|
|
await cfg.property.read_write()
|
|
await cfg.property.add('frozen')
|
|
await cfg.option('boolop').property.add('frozen')
|
|
prop = []
|
|
try:
|
|
await cfg.option('boolop').value.set([True])
|
|
except PropertiesOptionError as err:
|
|
prop = err.proptype
|
|
assert 'frozen' in prop
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_force_store_value():
|
|
descr = make_description_freeze()
|
|
async with await Config(descr) as cfg:
|
|
compare(await cfg.value.exportation(), (('wantref', 'wantref2', 'wantref3'), (None, None, None), (False, False, (False,)), ('forced', 'forced', 'forced')))
|
|
await cfg.option('wantref').value.set(True)
|
|
compare(await cfg.value.exportation(), (('wantref', 'wantref2', 'wantref3'), (None, None, None), (True, False, (False,)), ('user', 'forced', 'forced')))
|
|
await cfg.option('wantref').value.reset()
|
|
compare(await cfg.value.exportation(), (('wantref', 'wantref2', 'wantref3'), (None, None, None), (False, False, (False,)), ('forced', 'forced', 'forced')))
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_force_store_value_leadership_follower():
|
|
b = IntOption('int', 'Test int option', multi=True)
|
|
c = StrOption('str', 'Test string option', multi=True, properties=('force_store_value',))
|
|
descr = Leadership("int", "", [b, c])
|
|
with pytest.raises(ConfigError):
|
|
cfg = await Config(descr, session_id='error')
|
|
await delete_session('error')
|
|
assert not await list_sessions()
|
|
|
|
|
|
#@pytest.mark.asyncio
|
|
#async def test_force_store_value_leadership():
|
|
# b = IntOption('int', 'Test int option', multi=True, properties=('force_store_value',))
|
|
# c = StrOption('str', 'Test string option', multi=True)
|
|
# descr = Leadership("int", "", [b, c])
|
|
# cfg = await Config(descr)
|
|
# assert await cfg.value.get() == {'int': ('forced', ())}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_force_store_value_leadership_sub():
|
|
b = IntOption('int', 'Test int option', multi=True, properties=('force_store_value',))
|
|
c = StrOption('str', 'Test string option', multi=True)
|
|
descr = Leadership("int", "", [b, c])
|
|
odr = OptionDescription('odr', '', [descr])
|
|
async with await Config(odr) as cfg:
|
|
compare(await cfg.value.exportation(), (('int.int',), (None,), (tuple(),), ('forced',)))
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_force_store_value_callback():
|
|
b = IntOption('int', 'Test int option', Calculation(return_val), properties=('force_store_value',))
|
|
descr = OptionDescription("int", "", [b])
|
|
async with await Config(descr) as cfg:
|
|
compare(await cfg.value.exportation(), (('int',), (None,), (1,), ('forced',)))
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_force_store_value_callback_params():
|
|
b = IntOption('int', 'Test int option', Calculation(return_val2, Params(kwargs={'value': ParamValue(2)})), properties=('force_store_value',))
|
|
descr = OptionDescription("int", "", [b])
|
|
async with await Config(descr) as cfg:
|
|
compare(await cfg.value.exportation(), (('int',), (None,), (2,), ('forced',)))
|
|
assert not await list_sessions()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_force_store_value_callback_params_with_opt():
|
|
a = IntOption('val1', "", 2)
|
|
b = IntOption('int', 'Test int option', Calculation(return_val2, Params(kwargs={'value': ParamOption(a)})), properties=('force_store_value',))
|
|
descr = OptionDescription("int", "", [a, b])
|
|
async with await Config(descr) as cfg:
|
|
compare(await cfg.value.exportation(), (('int',), (None,), (2,), ('forced',)))
|
|
assert not await list_sessions()
|