tiramisu/tests/test_freeze.py

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()