Infra/tiramisu
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

New Postgres storage

Browse Source
This commit is contained in:
Emmanuel Garette
2020-01-22 20:46:18 +01:00
parent 746fa0134f
commit 3bef45c9db
85 changed files with 10814 additions and 9262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1360
tests/test_option_validator.py
View File

@ -1,6 +1,5 @@
from .autopath import do_autopath
do_autopath()
from .config import config_type, get_config
import warnings
import pytest
@ -13,10 +12,7 @@ from tiramisu.setting import groups
from tiramisu.error import ValueWarning, ConfigError, PropertiesOptionError
from tiramisu.i18n import _
from tiramisu.storage import list_sessions
def teardown_function(function):
assert list_sessions() == [], 'session list is not empty when leaving "{}"'.format(function.__name__)
from .config import config_type, get_config, event_loop
msg_err = _('attention, "{0}" could be an invalid {1} for "{2}"')
@ -88,35 +84,36 @@ async def test_validator(config_type):
opt1 = StrOption('opt1', '', validators=[Calculation(return_true, Params(ParamSelfOption()))], default='val')
opt2 = StrOption('opt2', '', validators=[Calculation(return_false, Params(ParamSelfOption()))])
root = OptionDescription('root', '', [opt1, opt2])
cfg_ori = await Config(root)
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == 'val'
with pytest.raises(ValueError):
await cfg.option('opt2').value.set('val')
try:
await cfg.option('opt2').value.set('val')
except ValueError as err:
msg = _('"{0}" is an invalid {1} for "{2}"').format('val', _('string'), 'opt2') + ', ' + _('test error return_false')
assert str(err) == msg
async with await Config(root) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == 'val'
with pytest.raises(ValueError):
await cfg.option('opt2').value.set('val')
try:
await cfg.option('opt2').value.set('val')
except ValueError as err:
msg = _('"{0}" is an invalid {1} for "{2}"').format('val', _('string'), 'opt2') + ', ' + _('test error return_false')
assert str(err) == msg
if config_type == 'tiramisu-api':
msg = _('"{0}" is an invalid {1} for "{2}"').format('val', 'string', 'opt2') + ', ' + _('test error return_false')
if config_type == 'tiramisu-api':
msg = _('"{0}" is an invalid {1} for "{2}"').format('val', 'string', 'opt2') + ', ' + _('test error return_false')
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
assert len(w) == 1
assert str(w[0].message) == msg
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.get()
assert len(w) == 1
assert str(w[0].message) == msg
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.get()
assert len(w) == 1
assert str(w[0].message) == msg
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
assert len(w) == 1
assert str(w[0].message) == msg
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.get()
assert len(w) == 1
assert str(w[0].message) == msg
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.get()
assert len(w) == 1
assert str(w[0].message) == msg
assert not await list_sessions()
@pytest.mark.asyncio
@ -124,39 +121,42 @@ async def test_validator_params(config_type):
opt1 = StrOption('opt1', '', validators=[Calculation(return_true, Params((ParamSelfOption(), ParamValue('yes'))))], default='val')
opt2 = StrOption('opt2', '', validators=[Calculation(return_false, Params((ParamSelfOption(), ParamValue('yes'))))])
root = OptionDescription('root', '', [opt1, opt2])
cfg_ori = await Config(root)
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == 'val'
with pytest.raises(ValueError):
await cfg.option('opt2').value.set('val')
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
assert len(w) == 1
async with await Config(root) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == 'val'
with pytest.raises(ValueError):
await cfg.option('opt2').value.set('val')
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
async def test_validator_params_value_values(config_type):
opt1 = StrOption('opt1', '', validators=[Calculation(value_values, Params((ParamSelfOption(whole=False), ParamSelfOption())))], default=['val'], multi=True)
root = OptionDescription('root', '', [opt1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('opt1').value.get() == ['val']
await cfg.option('opt1').value.set(['val1', 'val2'])
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('opt1').value.get() == ['val']
await cfg.option('opt1').value.set(['val1', 'val2'])
assert not await list_sessions()
@pytest.mark.asyncio
async def test_validator_params_value_values_index(config_type):
opt1 = StrOption('opt1', '', validators=[Calculation(value_values_index, Params((ParamSelfOption(whole=False), ParamSelfOption(), ParamIndex())))], default=['val'], multi=True)
root = OptionDescription('root', '', [opt1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('opt1').value.get() == ['val']
await cfg.option('opt1').value.set(['val1', 'val2'])
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('opt1').value.get() == ['val']
await cfg.option('opt1').value.set(['val1', 'val2'])
assert not await list_sessions()
@pytest.mark.asyncio
@ -165,10 +165,11 @@ async def test_validator_params_value_values_leader(config_type):
netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True)
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
root = OptionDescription('root', '', [interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val2'])
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val2'])
assert not await list_sessions()
@pytest.mark.asyncio
@ -177,10 +178,11 @@ async def test_validator_params_value_values_index_leader(config_type):
netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True)
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
root = OptionDescription('root', '', [interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val2'])
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val2'])
assert not await list_sessions()
@pytest.mark.asyncio
@ -189,13 +191,14 @@ async def test_validator_params_value_values_follower(config_type):
netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validators=[Calculation(value_values, Params((ParamSelfOption(), ParamSelfOption(whole=True))))])
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
root = OptionDescription('root', '', [interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val1'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2')
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val1'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2')
assert not await list_sessions()
@pytest.mark.asyncio
@ -204,13 +207,14 @@ async def test_validator_params_value_values_index_follower(config_type):
netmask_admin_eth0 = StrOption('netmask_admin_eth0', "masque du sous-reseau", multi=True, validators=[Calculation(value_values_index, Params((ParamSelfOption(), ParamSelfOption(whole=True), ParamIndex())))])
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
root = OptionDescription('root', '', [interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val1'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2')
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val1'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2')
assert not await list_sessions()
@pytest.mark.asyncio
@ -223,12 +227,13 @@ async def test_validator_params_value_values_kwargs_empty(config_type):
validators=[Calculation(value_empty, Params((ParamSelfOption(), ParamOption(v))))])
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
root = OptionDescription('root', '', [v, interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == ['ip']
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['ip', 'val'])
#await cfg.ip_admin_eth0.ip_admin_eth0.append('val')
#await cfg.ip_admin_eth0.netmask_admin_eth0[1] = 'val2'
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == ['ip']
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['ip', 'val'])
#await cfg.ip_admin_eth0.ip_admin_eth0.append('val')
#await cfg.ip_admin_eth0.netmask_admin_eth0[1] = 'val2'
assert not await list_sessions()
@pytest.mark.asyncio
@ -241,12 +246,13 @@ async def test_validator_params_value_values_kwargs(config_type):
validators=[Calculation(value_values_auto, Params((ParamSelfOption(), ParamSelfOption(whole=True)), kwargs={'auto': ParamOption(v)}))])
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
root = OptionDescription('root', '', [v, interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == ['ip']
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['ip', 'val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2')
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == ['ip']
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['ip', 'val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2')
assert not await list_sessions()
@pytest.mark.asyncio
@ -258,13 +264,14 @@ async def test_validator_params_value_values_kwargs_values(config_type):
validators=[Calculation(value_values_auto2, Params(ParamSelfOption(), kwargs={'values': ParamOption(ip_admin_eth0)}))])
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
root = OptionDescription('root', '', [interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2')
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val2')
assert not await list_sessions()
@pytest.mark.asyncio
@ -272,41 +279,43 @@ async def test_validator_params_option(config_type):
opt0 = StrOption('opt0', '', default='yes')
opt1 = StrOption('opt1', '', validators=[Calculation(return_true, Params((ParamSelfOption(), ParamOption(opt0))))], default='val')
r = OptionDescription('root', '', [opt0, opt1])
cfg_ori = await Config(r)
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == 'val'
await cfg.option('opt0').value.set('val')
with pytest.raises(ValueError):
await cfg.option('opt1').value.get()
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt1').value.get()
assert len(w) == 1
async with await Config(r) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == 'val'
await cfg.option('opt0').value.set('val')
with pytest.raises(ValueError):
await cfg.option('opt1').value.get()
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt1').value.get()
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
async def test_validator_multi(config_type):
opt1 = StrOption('opt1', '', validators=[Calculation(return_if_val, Params(ParamSelfOption(whole=False)))], multi=True)
root = OptionDescription('root', '', [opt1])
cfg_ori = await Config(root)
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == []
await cfg.option('opt1').value.set(['val'])
assert await cfg.option('opt1').value.get() == ['val']
with pytest.raises(ValueError):
await cfg.option('opt1').value.set(['val', 'val1'])
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt1').value.set(['val', 'val1'])
assert len(w) == 1
async with await Config(root) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == []
await cfg.option('opt1').value.set(['val'])
assert await cfg.option('opt1').value.get() == ['val']
with pytest.raises(ValueError):
await cfg.option('opt1').value.set(['val', 'val1'])
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt1').value.set(['val', 'val1'])
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -315,46 +324,47 @@ async def test_validator_warning(config_type):
opt2 = StrOption('opt2', '', validators=[Calculation(return_false, Params(ParamSelfOption()), warnings_only=True)])
opt3 = StrOption('opt3', '', validators=[Calculation(return_if_val, Params(ParamSelfOption(whole=False)), warnings_only=True)], multi=True, properties=('notunique',))
root = OptionDescription('root', '', [opt1, opt2, opt3])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('opt1').value.get() == 'val'
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt1').value.set('val')
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
assert len(w) == 1
if config_type != 'tiramisu-api':
assert w[0].message.opt() == opt2
assert str(w[0].message) == msg_err.format('val', opt2._display_name, 'opt2') + ', ' + 'test error return_false'
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt3').value.set(['val'])
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt3').value.set(['val', 'val1'])
assert len(w) == 1
if config_type != 'tiramisu-api':
assert w[0].message.opt() == opt3
assert str(w[0].message) == msg_err.format('val1', opt3._display_name, 'opt3') + ', ' + 'test error'
#
with warnings.catch_warnings(record=True) as w:
with pytest.raises(ValueError):
await cfg.option('opt2').value.set(1)
assert len(w) == 0
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
await cfg.option('opt3').value.set(['val', 'val1', 'val'])
assert len(w) == 2
if config_type != 'tiramisu-api':
assert w[0].message.opt() == opt2
assert str(w[0].message) == msg_err.format('val', opt2._display_name, 'opt2') + ', ' + 'test error return_false'
assert w[1].message.opt() == opt3
assert str(w[1].message) == msg_err.format('val1', opt3._display_name, 'opt3') + ', ' + 'test error'
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('opt1').value.get() == 'val'
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt1').value.set('val')
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
assert len(w) == 1
if config_type != 'tiramisu-api':
assert w[0].message.opt() == opt2
assert str(w[0].message) == msg_err.format('val', opt2._display_name, 'opt2') + ', ' + 'test error return_false'
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt3').value.set(['val'])
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt3').value.set(['val', 'val1'])
assert len(w) == 1
if config_type != 'tiramisu-api':
assert w[0].message.opt() == opt3
assert str(w[0].message) == msg_err.format('val1', opt3._display_name, 'opt3') + ', ' + 'test error'
#
with warnings.catch_warnings(record=True) as w:
with pytest.raises(ValueError):
await cfg.option('opt2').value.set(1)
assert len(w) == 0
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
await cfg.option('opt3').value.set(['val', 'val1', 'val'])
assert len(w) == 2
if config_type != 'tiramisu-api':
assert w[0].message.opt() == opt2
assert str(w[0].message) == msg_err.format('val', opt2._display_name, 'opt2') + ', ' + 'test error return_false'
assert w[1].message.opt() == opt3
assert str(w[1].message) == msg_err.format('val1', opt3._display_name, 'opt3') + ', ' + 'test error'
assert not await list_sessions()
@pytest.mark.asyncio
@ -363,43 +373,44 @@ async def test_validator_warning_disabled(config_type):
opt2 = StrOption('opt2', '', validators=[Calculation(return_false, Params(ParamSelfOption()), warnings_only=True)])
opt3 = StrOption('opt3', '', validators=[Calculation(return_if_val, Params(ParamSelfOption(whole=False)), warnings_only=True)], multi=True, properties=('notunique',))
root = OptionDescription('root', '', [opt1, opt2, opt3])
cfg_ori = await Config(root)
await cfg_ori.property.pop('warnings')
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == 'val'
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt1').value.set('val')
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt3').value.set(['val'])
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt3').value.set(['val', 'val1'])
assert w == []
with pytest.raises(ValueError):
await cfg.option('opt2').value.set(1)
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
await cfg.option('opt3').value.set(['val', 'val1', 'val'])
assert w == []
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
if config_type != 'tiramisu-api':
async with await Config(root) as cfg_ori:
await cfg_ori.property.pop('warnings')
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('opt1').value.get() == 'val'
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt1').value.set('val')
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt3').value.set(['val'])
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt3').value.set(['val', 'val1'])
assert w == []
with pytest.raises(ValueError):
await cfg.option('opt2').value.set(1)
assert len(w) == 1
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set('val')
await cfg.option('opt3').value.set(['val', 'val1', 'val'])
assert w == []
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
if config_type != 'tiramisu-api':
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('opt2').value.set(1)
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -411,52 +422,53 @@ async def test_validator_warning_leadership(config_type):
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
assert interface1.impl_get_group_type() == groups.leadership
root = OptionDescription('root', '', [interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set([None])
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
assert len(w) == 1
if config_type != 'tiramisu-api':
assert w[0].message.opt() == netmask_admin_eth0
assert str(w[0].message) == msg_err.format('val1', netmask_admin_eth0._display_name, display_name_netmask) + ', test error'
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val'])
if config_type != 'tiramisu-api':
assert w[0].message.opt() == ip_admin_eth0
assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip) + ', test error return_false'
else:
assert len(w) == 2
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val1', 'val1'])
if config_type != 'tiramisu-api':
assert w[0].message.opt() == ip_admin_eth0
assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip) + ', test error return_false'
else:
assert len(w) == 3
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val', 'val1'])
if config_type != 'tiramisu-api':
assert w[0].message.opt() == ip_admin_eth0
assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip) + ', test error return_false'
else:
assert len(w) == 3
#
warnings.resetwarnings()
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val1', 'val'])
if config_type != 'tiramisu-api':
assert w[0].message.opt() == ip_admin_eth0
assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip) + ', test error return_false'
else:
assert len(w) == 3
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set([None])
assert w == []
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val1')
assert len(w) == 1
if config_type != 'tiramisu-api':
assert w[0].message.opt() == netmask_admin_eth0
assert str(w[0].message) == msg_err.format('val1', netmask_admin_eth0._display_name, display_name_netmask) + ', test error'
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val'])
if config_type != 'tiramisu-api':
assert w[0].message.opt() == ip_admin_eth0
assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip) + ', test error return_false'
else:
assert len(w) == 2
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val', 'val1', 'val1'])
if config_type != 'tiramisu-api':
assert w[0].message.opt() == ip_admin_eth0
assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip) + ', test error return_false'
else:
assert len(w) == 3
#
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val', 'val1'])
if config_type != 'tiramisu-api':
assert w[0].message.opt() == ip_admin_eth0
assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip) + ', test error return_false'
else:
assert len(w) == 3
#
warnings.resetwarnings()
with warnings.catch_warnings(record=True) as w:
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['val1', 'val1', 'val'])
if config_type != 'tiramisu-api':
assert w[0].message.opt() == ip_admin_eth0
assert str(w[0].message) == msg_err.format('val', ip_admin_eth0._display_name, display_name_ip) + ', test error return_false'
else:
assert len(w) == 3
assert not await list_sessions()
@pytest.mark.asyncio
@ -468,13 +480,14 @@ async def test_validator_follower_param(config_type):
validators=[Calculation(return_true, Params(ParamSelfOption(), kwargs={'param': ParamOption(ip_admin_eth0)}))])
interface1 = Leadership('ip_admin_eth0', '', [ip_admin_eth0, netmask_admin_eth0])
root = OptionDescription('root', '', [interface1])
cfg = await Config(root)
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['yes'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['yes', 'yes'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val')
async with await Config(root) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('ip_admin_eth0.ip_admin_eth0').value.get() == []
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['yes'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 0).value.set('val')
await cfg.option('ip_admin_eth0.ip_admin_eth0').value.set(['yes', 'yes'])
await cfg.option('ip_admin_eth0.netmask_admin_eth0', 1).value.set('val')
assert not await list_sessions()
@pytest.mark.asyncio
@ -485,14 +498,15 @@ async def test_validator_dependencies():
validators=[Calculation(return_true, Params(ParamSelfOption(whole=False), kwargs={'param': ParamOption(ip_admin_eth0)}))])
opt2 = StrOption('opt2', '', validators=[Calculation(return_false, Params(ParamSelfOption(whole=False)))])
root = OptionDescription('root', '', [ip_admin_eth0, netmask_admin_eth0, opt2])
cfg = await Config(root)
assert await cfg.option('ip_admin_eth0').option.has_dependency() is False
assert await cfg.option('netmask_admin_eth0').option.has_dependency() is True
assert await cfg.option('opt2').option.has_dependency() is False
#
assert await cfg.option('ip_admin_eth0').option.has_dependency(False) is True
assert await cfg.option('netmask_admin_eth0').option.has_dependency(False) is False
assert await cfg.option('opt2').option.has_dependency(False) is False
async with await Config(root) as cfg:
assert await cfg.option('ip_admin_eth0').option.has_dependency() is False
assert await cfg.option('netmask_admin_eth0').option.has_dependency() is True
assert await cfg.option('opt2').option.has_dependency() is False
#
assert await cfg.option('ip_admin_eth0').option.has_dependency(False) is True
assert await cfg.option('netmask_admin_eth0').option.has_dependency(False) is False
assert await cfg.option('opt2').option.has_dependency(False) is False
assert not await list_sessions()
@pytest.mark.asyncio
@ -500,33 +514,34 @@ async def test_validator_ip_netmask(config_type):
a = IPOption('a', '')
b = NetmaskOption('b', '', validators=[Calculation(valid_ip_netmask, Params((ParamOption(a, todict=True), ParamSelfOption())))])
od = OptionDescription('od', '', [a, b])
cfg_ori = await Config(od)
cfg = cfg_ori
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a').value.set('192.168.1.1')
await cfg.option('b').value.set('255.255.255.0')
await cfg.option('a').value.set('192.168.1.2')
await cfg.option('b').value.set('255.255.255.128')
await cfg.option('b').value.set('255.255.255.0')
await cfg.option('a').value.set('192.168.1.0')
with pytest.raises(ValueError):
await cfg.option('b').value.get()
await cfg.option('a').value.set('192.168.1.255')
with pytest.raises(ValueError):
await cfg.option('b').value.get()
await cfg.option('a').value.reset()
await cfg.option('b').value.reset()
await cfg.option('a').value.set('192.168.1.255')
with pytest.raises(ValueError):
async with await Config(od) as cfg_ori:
cfg = cfg_ori
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a').value.set('192.168.1.1')
await cfg.option('b').value.set('255.255.255.0')
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('a').value.set('192.168.1.2')
await cfg.option('b').value.set('255.255.255.128')
await cfg.option('b').value.set('255.255.255.0')
assert len(w) == 1
await cfg.option('a').value.set('192.168.1.0')
with pytest.raises(ValueError):
await cfg.option('b').value.get()
await cfg.option('a').value.set('192.168.1.255')
with pytest.raises(ValueError):
await cfg.option('b').value.get()
await cfg.option('a').value.reset()
await cfg.option('b').value.reset()
await cfg.option('a').value.set('192.168.1.255')
with pytest.raises(ValueError):
await cfg.option('b').value.set('255.255.255.0')
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('b').value.set('255.255.255.0')
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -534,27 +549,28 @@ async def test_validator_network_netmask(config_type):
a = NetworkOption('a', '')
b = NetmaskOption('b', '', validators=[Calculation(valid_network_netmask, Params((ParamOption(a, todict=True), ParamSelfOption())))])
od = OptionDescription('od', '', [a, b])
cfg_ori = await Config(od)
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a').value.set('192.168.1.1')
await cfg.option('b').value.set('255.255.255.255')
await cfg.option('b').value.reset()
await cfg.option('a').value.set('192.168.1.0')
await cfg.option('b').value.set('255.255.255.0')
await cfg.option('a').value.set('192.168.1.1')
with pytest.raises(ValueError):
await cfg.option('b').value.get()
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
async with await Config(od) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a').value.set('192.168.1.1')
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
await cfg.option('b').value.get()
assert len(w) == 1
await cfg.option('b').value.set('255.255.255.255')
await cfg.option('b').value.reset()
await cfg.option('a').value.set('192.168.1.0')
await cfg.option('b').value.set('255.255.255.0')
await cfg.option('a').value.set('192.168.1.1')
with pytest.raises(ValueError):
await cfg.option('b').value.get()
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('a').value.set('192.168.1.1')
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
await cfg.option('b').value.get()
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -565,20 +581,21 @@ async def test_validator_ip_in_network(config_type):
d = IPOption('d', '', validators=[Calculation(valid_in_network, Params((ParamSelfOption(), ParamOption(a, todict=True), ParamOption(b, todict=True))), warnings_only=True)])
od = OptionDescription('od', '', [a, b, c, d])
warnings.simplefilter("always", ValueWarning)
cfg = await Config(od)
cfg = await get_config(cfg, config_type)
await cfg.option('a').value.set('192.168.1.0')
await cfg.option('b').value.set('255.255.255.0')
await cfg.option('c').value.set('192.168.1.1')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.2.1')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.1.0')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.1.255')
with warnings.catch_warnings(record=True) as w:
await cfg.option('d').value.set('192.168.2.1')
assert len(w) == 1
async with await Config(od) as cfg:
cfg = await get_config(cfg, config_type)
await cfg.option('a').value.set('192.168.1.0')
await cfg.option('b').value.set('255.255.255.0')
await cfg.option('c').value.set('192.168.1.1')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.2.1')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.1.0')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.1.255')
with warnings.catch_warnings(record=True) as w:
await cfg.option('d').value.set('192.168.2.1')
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -588,19 +605,20 @@ async def test_validator_ip_in_network_cidr(config_type):
d = IPOption('d', '', validators=[Calculation(valid_in_network, Params((ParamSelfOption(), ParamOption(a, todict=True))), warnings_only=True)])
od = OptionDescription('od', '', [a, c, d])
warnings.simplefilter("always", ValueWarning)
cfg = await Config(od)
cfg = await get_config(cfg, config_type)
await cfg.option('a').value.set('192.168.1.0/24')
await cfg.option('c').value.set('192.168.1.1')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.2.1')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.1.0')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.1.255')
with warnings.catch_warnings(record=True) as w:
await cfg.option('d').value.set('192.168.2.1')
assert len(w) == 1
async with await Config(od) as cfg:
cfg = await get_config(cfg, config_type)
await cfg.option('a').value.set('192.168.1.0/24')
await cfg.option('c').value.set('192.168.1.1')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.2.1')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.1.0')
with pytest.raises(ValueError):
await cfg.option('c').value.set('192.168.1.255')
with warnings.catch_warnings(record=True) as w:
await cfg.option('d').value.set('192.168.2.1')
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -609,26 +627,27 @@ async def test_validator_ip_netmask_multi(config_type):
b = NetmaskOption('b', '', multi=True, validators=[Calculation(valid_ip_netmask, Params((ParamOption(a, todict=True), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od2', '', [od])
cfg_ori = await Config(od2)
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a.a').value.set(['192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.2'])
await cfg.option('a.b', 0).value.set('255.255.255.128')
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.0'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
#
await cfg.option('a.a').value.set(['192.168.1.2'])
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a.a').value.set(['192.168.1.0'])
with warnings.catch_warnings(record=True) as w:
await cfg.option('a.b', 0).value.get()
assert len(w) == 1
async with await Config(od2) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a.a').value.set(['192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.2'])
await cfg.option('a.b', 0).value.set('255.255.255.128')
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.0'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
#
await cfg.option('a.a').value.set(['192.168.1.2'])
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a.a').value.set(['192.168.1.0'])
with warnings.catch_warnings(record=True) as w:
await cfg.option('a.b', 0).value.get()
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -637,16 +656,17 @@ async def test_validator_network_netmask_multi(config_type):
b = NetmaskOption('b', '', multi=True, validators=[Calculation(valid_network_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od', '', [od])
cfg = await Config(od2)
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set(['192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.255')
await cfg.option('a.b', 0).value.reset()
await cfg.option('a.a').value.set(['192.168.1.0'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
async with await Config(od2) as cfg:
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set(['192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.255')
await cfg.option('a.b', 0).value.reset()
await cfg.option('a.a').value.set(['192.168.1.0'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
assert not await list_sessions()
@pytest.mark.asyncio
@ -655,12 +675,13 @@ async def test_validator_network_netmask_multi_follower_default_multi(config_typ
b = NetmaskOption('b', '', default_multi=u'255.255.255.0', multi=True, properties=('mandatory',), validators=[Calculation(valid_network_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od2', '', [od])
cfg = await Config(od2)
await cfg.property.read_write()
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set([undefined])
assert await cfg.option('a.a').value.get() == ['192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
async with await Config(od2) as cfg:
await cfg.property.read_write()
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set([undefined])
assert await cfg.option('a.a').value.get() == ['192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
assert not await list_sessions()
@pytest.mark.asyncio
@ -669,31 +690,32 @@ async def test_validator_network_netmask_multi_follower_default(config_type):
b = NetmaskOption('b', '', default_multi=u'255.255.255.0', multi=True, properties=('mandatory',), validators=[Calculation(valid_network_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od2', '', [od])
cfg_ori = await Config(od2)
await cfg_ori.property.read_write()
await cfg_ori.property.pop('cache')
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set(['192.168.1.0'])
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('a.a').value.get() == [u'192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == u'255.255.255.0'
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('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.set([u'192.168.1.0'])
with pytest.raises(ValueError):
await cfg.option('a.b', 1).value.set([u'192.168.1.1'])
await cfg.option('a.a').value.set(['192.168.1.0', undefined])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.b', 1).value.set('255.255.255.255')
await cfg.option('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])
async with await Config(od2) as cfg_ori:
await cfg_ori.property.read_write()
await cfg_ori.property.pop('cache')
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set(['192.168.1.0'])
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('a.a').value.get() == [u'192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == u'255.255.255.0'
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('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.set([u'192.168.1.0'])
with pytest.raises(ValueError):
await cfg.option('a.b', 1).value.set([u'192.168.1.1'])
await cfg.option('a.a').value.set(['192.168.1.0', undefined])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.b', 1).value.set('255.255.255.255')
await cfg.option('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])
assert not await list_sessions()
def return_netmask(*args, **kwargs):
@ -715,30 +737,31 @@ async def test_validator_network_netmask_multi_follower_callback(config_type):
b = NetmaskOption('b', '', Calculation(return_netmask, Params(kwargs={'index': ParamIndex()})), multi=True, properties=('mandatory',), validators=[Calculation(valid_network_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od2', '', [od])
cfg_ori = await Config(od2)
await cfg_ori.property.read_write()
await cfg_ori.property.pop('cache')
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set(['192.168.1.0'])
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('a.a').value.get() == [u'192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
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('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])
await cfg.option('a.b', 0).value.get()
with pytest.raises(ValueError):
await cfg.option('a.b', 1).value.get()
await cfg.option('a.a').value.set(['192.168.1.0', undefined])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.b', 1).value.set('255.255.255.255')
await cfg.option('a.a').value.set(['192.168.1.0', '192.168.1.1'])
async with await Config(od2) as cfg_ori:
await cfg_ori.property.read_write()
await cfg_ori.property.pop('cache')
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set(['192.168.1.0'])
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('a.a').value.get() == [u'192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
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('a.a').value.set([u'192.168.1.0', u'192.168.1.1'])
await cfg.option('a.b', 0).value.get()
with pytest.raises(ValueError):
await cfg.option('a.b', 1).value.get()
await cfg.option('a.a').value.set(['192.168.1.0', undefined])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.b', 1).value.set('255.255.255.255')
await cfg.option('a.a').value.set(['192.168.1.0', '192.168.1.1'])
assert not await list_sessions()
@pytest.mark.asyncio
@ -747,32 +770,33 @@ async def test_validator_network_netmask_multi_follower_callback_value(config_ty
b = NetmaskOption('b', '', Calculation(return_netmask2, Params(ParamOption(a))), multi=True, properties=('mandatory',), validators=[Calculation(valid_network_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od2', '', [od])
cfg = await Config(od2)
await cfg.property.read_write()
await cfg.property.pop('cache')
cfg = await get_config(cfg, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set(['192.168.1.0'])
assert await cfg.option('a.a').value.get() == ['192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
await cfg.option('a.a').value.set(['192.168.1.0', '192.168.2.1'])
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
with pytest.raises(ValueError):
await cfg.option('a.b', 1).value.get()
await cfg.option('a.a').value.pop(1)
#
assert await cfg.option('a.a').value.get() == [u'192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
await cfg.option('a.a').value.set(['192.168.2.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
await cfg.option('a.a').value.set(['192.168.1.0'])
#
assert await cfg.option('a.a').value.get() == [u'192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
await cfg.option('a.a').value.set(['192.168.1.0', '192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.b', 1).value.set('255.255.255.255')
async with await Config(od2) as cfg:
await cfg.property.read_write()
await cfg.property.pop('cache')
cfg = await get_config(cfg, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set(['192.168.1.0'])
assert await cfg.option('a.a').value.get() == ['192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
await cfg.option('a.a').value.set(['192.168.1.0', '192.168.2.1'])
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
with pytest.raises(ValueError):
await cfg.option('a.b', 1).value.get()
await cfg.option('a.a').value.pop(1)
#
assert await cfg.option('a.a').value.get() == [u'192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
await cfg.option('a.a').value.set(['192.168.2.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
await cfg.option('a.a').value.set(['192.168.1.0'])
#
assert await cfg.option('a.a').value.get() == [u'192.168.1.0']
assert await cfg.option('a.b', 0).value.get() == '255.255.255.0'
await cfg.option('a.a').value.set(['192.168.1.0', '192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.b', 1).value.set('255.255.255.255')
assert not await list_sessions()
@pytest.mark.asyncio
@ -781,20 +805,21 @@ async def test_validator_ip_netmask_multi_leader(config_type):
b = NetmaskOption('b', '', multi=True, validators=[Calculation(valid_ip_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od2', '', [od])
cfg = await Config(od2)
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set(['192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.2'])
await cfg.option('a.b', 0).value.set('255.255.255.128')
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.0'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
await cfg.option('a.a').value.set(['192.168.1.128'])
with pytest.raises(ValueError):
async with await Config(od2) as cfg:
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set(['192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.2'])
await cfg.option('a.b', 0).value.set('255.255.255.128')
await cfg.option('a.a').value.set(['192.168.1.2', '192.168.1.3'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.0'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
await cfg.option('a.a').value.set(['192.168.1.128'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.set('255.255.255.128')
await cfg.option('a.a').value.set(['192.168.1.2', '192.168.1.3'])
assert not await list_sessions()
@pytest.mark.asyncio
@ -803,16 +828,17 @@ async def test_validator_network_netmask_multi_leader(config_type):
b = NetmaskOption('b', '', multi=True, validators=[Calculation(valid_network_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od2', '', [od])
cfg = await Config(od2)
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set(['192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.255')
await cfg.option('a.b', 0).value.reset()
await cfg.option('a.a').value.set(['192.168.1.0'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
async with await Config(od2) as cfg:
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set(['192.168.1.1'])
await cfg.option('a.b', 0).value.set('255.255.255.255')
await cfg.option('a.b', 0).value.reset()
await cfg.option('a.a').value.set(['192.168.1.0'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.a').value.set(['192.168.1.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
assert not await list_sessions()
@pytest.mark.asyncio
@ -822,30 +848,31 @@ async def test_validator_broadcast(config_type):
c = BroadcastOption('c', '', multi=True, validators=[Calculation(valid_broadcast, Params((ParamOption(a), ParamOption(b), ParamSelfOption())))])
od = Leadership('a', '', [a, b, c])
od2 = OptionDescription('od2', '', [od])
cfg = await Config(od2)
cfg = await get_config(cfg, config_type)
#first, test network_netmask
await cfg.option('a.a').value.set(['192.168.1.128'])
with pytest.raises(ValueError):
await cfg.option('a.a').value.set(['255.255.255.0'])
#
await cfg.option('a.a').value.set(['192.168.1.0'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.c', 0).value.set('192.168.1.255')
await cfg.option('a.a').value.set(['192.168.1.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
with pytest.raises(ValueError):
await cfg.option('a.c', 0).value.get()
#
await cfg.option('a.a').value.set(['192.168.1.0', '192.168.2.128'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.b', 1).value.set('255.255.255.128')
await cfg.option('a.c', 0).value.set('192.168.1.255')
await cfg.option('a.c', 1).value.set('192.168.2.255')
with pytest.raises(ValueError):
await cfg.option('a.c', 1).value.set('192.168.2.128')
await cfg.option('a.c', 1).value.set('192.168.2.255')
async with await Config(od2) as cfg:
cfg = await get_config(cfg, config_type)
#first, test network_netmask
await cfg.option('a.a').value.set(['192.168.1.128'])
with pytest.raises(ValueError):
await cfg.option('a.a').value.set(['255.255.255.0'])
#
await cfg.option('a.a').value.set(['192.168.1.0'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.c', 0).value.set('192.168.1.255')
await cfg.option('a.a').value.set(['192.168.1.1'])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
with pytest.raises(ValueError):
await cfg.option('a.c', 0).value.get()
#
await cfg.option('a.a').value.set(['192.168.1.0', '192.168.2.128'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.b', 1).value.set('255.255.255.128')
await cfg.option('a.c', 0).value.set('192.168.1.255')
await cfg.option('a.c', 1).value.set('192.168.2.255')
with pytest.raises(ValueError):
await cfg.option('a.c', 1).value.set('192.168.2.128')
await cfg.option('a.c', 1).value.set('192.168.2.255')
assert not await list_sessions()
@pytest.mark.asyncio
@ -854,19 +881,20 @@ async def test_validator_broadcast_warnings(config_type):
a = NetworkOption('a', '', properties=('mandatory', 'disabled'))
b = NetmaskOption('b', '', properties=('mandatory', 'disabled'), validators=[Calculation(valid_network_netmask, Params((ParamOption(a), ParamSelfOption())), warnings_only=True)])
od = OptionDescription('a', '', [a, b])
cfg_ori = await Config(od)
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('a').value.set('192.168.1.4')
await cfg.option('b').value.set('255.255.255.0')
assert len(w) == 1
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.read_write()
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
list(await cfg.value.mandatory())
assert len(w) == 0
async with await Config(od) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('a').value.set('192.168.1.4')
await cfg.option('b').value.set('255.255.255.0')
assert len(w) == 1
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.read_write()
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
list(await cfg.value.mandatory())
assert len(w) == 0
assert not await list_sessions()
@pytest.mark.asyncio
@ -875,9 +903,10 @@ async def test_validator_broadcast_default_1():
b = NetmaskOption('b', '', '255.255.255.128')
c = BroadcastOption('c', '', '192.168.2.127', validators=[Calculation(valid_broadcast, Params((ParamOption(a), ParamOption(b), ParamSelfOption())))])
od = OptionDescription('a', '', [a, b, c])
cfg = await Config(od)
with pytest.raises(ValueError):
await cfg.value.dict()
async with await Config(od) as cfg:
with pytest.raises(ValueError):
await cfg.value.dict()
assert not await list_sessions()
@pytest.mark.asyncio
@ -886,8 +915,9 @@ async def test_validator_broadcast_default_2():
b = NetmaskOption('b', '', '255.255.255.128')
d = BroadcastOption('d', '', '192.168.1.127', validators=[Calculation(valid_broadcast, Params((ParamOption(a), ParamOption(b), ParamSelfOption())))])
od = OptionDescription('a', '', [a, b, d])
cfg = await Config(od)
assert await cfg.value.dict()
async with await Config(od) as cfg:
assert await cfg.value.dict()
assert not await list_sessions()
@pytest.mark.asyncio
@ -897,11 +927,12 @@ async def test_validator_not_all(config_type):
c = BroadcastOption('c', '', multi=True)
od = Leadership('a', '', [a, b, c])
od = OptionDescription('od2', '', [od])
cfg = await Config(od)
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set(['192.168.1.0'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.c', 0).value.set('192.168.1.255')
async with await Config(od) as cfg:
cfg = await get_config(cfg, config_type)
await cfg.option('a.a').value.set(['192.168.1.0'])
await cfg.option('a.b', 0).value.set('255.255.255.0')
await cfg.option('a.c', 0).value.set('192.168.1.255')
assert not await list_sessions()
@pytest.mark.asyncio
@ -910,10 +941,11 @@ async def test_validator_network_netmask_mandatory(config_type):
b = NetmaskOption('b', '', multi=True, properties=('mandatory',), default_multi=u'0.0.0.0', validators=[Calculation(valid_network_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('od2', '', [od])
cfg = await Config(od2)
await cfg.property.read_only()
cfg = await get_config(cfg, config_type)
await cfg.value.dict()
async with await Config(od2) as cfg:
await cfg.property.read_only()
cfg = await get_config(cfg, config_type)
await cfg.value.dict()
assert not await list_sessions()
@pytest.mark.asyncio
@ -921,11 +953,12 @@ async def test_validator_has_dependency():
a = IPOption('a', '')
b = NetmaskOption('b', '', validators=[Calculation(valid_ip_netmask, Params((ParamOption(a), ParamSelfOption())))])
od = OptionDescription('od', '', [a, b])
cfg = await Config(od)
assert await cfg.option('a').option.has_dependency() is False
assert await cfg.option('b').option.has_dependency() is True
assert await cfg.option('a').option.has_dependency(False) is True
assert await cfg.option('b').option.has_dependency(False) is False
async with await Config(od) as cfg:
assert await cfg.option('a').option.has_dependency() is False
assert await cfg.option('b').option.has_dependency() is True
assert await cfg.option('a').option.has_dependency(False) is True
assert await cfg.option('b').option.has_dependency(False) is False
assert not await list_sessions()
@pytest.mark.asyncio
@ -934,18 +967,19 @@ async def test_validator_warnings_only_more_option(config_type):
b = IntOption('b', '')
d = IntOption('d', '', validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a, todict=True), ParamOption(b, todict=True))), warnings_only=True)])
od = OptionDescription('od', '', [a, b, d])
cfg = await Config(od)
cfg = await get_config(cfg, config_type)
await cfg.option('a').value.set(1)
await cfg.option('b').value.set(1)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('d').value.get()
assert w == []
with warnings.catch_warnings(record=True) as w:
await cfg.option('d').value.set(1)
assert w != []
assert len(w) == 1
async with await Config(od) as cfg:
cfg = await get_config(cfg, config_type)
await cfg.option('a').value.set(1)
await cfg.option('b').value.set(1)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('d').value.get()
assert w == []
with warnings.catch_warnings(record=True) as w:
await cfg.option('d').value.set(1)
assert w != []
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -953,17 +987,18 @@ async def test_validator_error_prefix():
a = IntOption('a', '')
b = IntOption('b', '', validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a, todict=True))))])
od = OptionDescription('od', '', [a, b])
cfg = await Config(od)
await cfg.option('a').value.set(1)
try:
await cfg.option('b').value.set(1)
except Exception as err:
assert str(err) == _('"{0}" is an invalid {1} for "{2}"').format('1', _('integer'), 'b') + ', ' + _('value is identical to {}').format('"a"')
try:
await cfg.option('b').value.set(1)
except Exception as err:
err.prefix = ''
assert str(err) == _('value is identical to {}').format('"a"')
async with await Config(od) as cfg:
await cfg.option('a').value.set(1)
try:
await cfg.option('b').value.set(1)
except Exception as err:
assert str(err) == _('"{0}" is an invalid {1} for "{2}"').format('1', _('integer'), 'b') + ', ' + _('value is identical to {}').format('"a"')
try:
await cfg.option('b').value.set(1)
except Exception as err:
err.prefix = ''
assert str(err) == _('value is identical to {}').format('"a"')
assert not await list_sessions()
@pytest.mark.asyncio
@ -971,11 +1006,12 @@ async def test_validator_warnings_only_option(config_type):
a = IntOption('a', '')
b = IntOption('b', '', warnings_only=True, validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a, todict=True))))])
od = OptionDescription('od', '', [a, b])
cfg_ori = await Config(od)
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a').value.set(1)
with pytest.raises(ValueError):
await cfg.option('b').value.set(1)
async with await Config(od) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
await cfg.option('a').value.set(1)
with pytest.raises(ValueError):
await cfg.option('b').value.set(1)
assert not await list_sessions()
@pytest.mark.asyncio
@ -983,25 +1019,26 @@ async def test_validator_not_equal(config_type):
a = IntOption('a', '')
b = IntOption('b', '', validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a))))])
od = OptionDescription('od', '', [a, b])
cfg_ori = await Config(od)
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('a').value.get() is None
assert await cfg.option('b').value.get() is None
await cfg.option('a').value.set(1)
await cfg.option('a').value.reset()
await cfg.option('a').value.set(1)
with pytest.raises(ValueError):
await cfg.option('b').value.set(1)
await cfg.option('b').value.set(2)
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('b').value.set(1)
assert len(w) == 1
async with await Config(od) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
assert await cfg.option('a').value.get() is None
assert await cfg.option('b').value.get() is None
await cfg.option('a').value.set(1)
await cfg.option('a').value.reset()
await cfg.option('a').value.set(1)
with pytest.raises(ValueError):
await cfg.option('b').value.set(1)
await cfg.option('b').value.set(2)
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.add('demoting_error_warning')
cfg = await get_config(cfg_ori, config_type)
warnings.simplefilter("always", ValueWarning)
with warnings.catch_warnings(record=True) as w:
await cfg.option('b').value.set(1)
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -1010,18 +1047,19 @@ async def test_validator_not_equal_leadership(config_type):
b = IntOption('b', '', multi=True, validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a))))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('b', '', [od])
cfg = await Config(od2)
cfg = await get_config(cfg, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set([1])
await cfg.option('a.a').value.reset()
await cfg.option('a.a').value.set([1])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.set(1)
await cfg.option('a.b', 0).value.set(2)
await cfg.option('a.a').value.reset()
await cfg.option('a.a').value.set([1])
await cfg.value.dict()
async with await Config(od2) as cfg:
cfg = await get_config(cfg, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set([1])
await cfg.option('a.a').value.reset()
await cfg.option('a.a').value.set([1])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.set(1)
await cfg.option('a.b', 0).value.set(2)
await cfg.option('a.a').value.reset()
await cfg.option('a.a').value.set([1])
await cfg.value.dict()
assert not await list_sessions()
@pytest.mark.asyncio
@ -1030,20 +1068,21 @@ async def test_validator_not_equal_leadership_default():
b = IntOption('b', '', multi=True, default_multi=1, validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a))))])
od = Leadership('a', '', [a, b])
od2 = OptionDescription('a', '', [od])
cfg = await Config(od2)
# FIXME cfg = await get_config(cfg, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set([1])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
await cfg.option('a.a').value.set([2])
await cfg.option('a.a').value.reset()
await cfg.option('a.a').value.set([2])
#
await cfg.property.add('demoting_error_warning')
with warnings.catch_warnings(record=True) as w:
await cfg.option('a.b', 0).value.set(2)
assert len(w) == 1
async with await Config(od2) as cfg:
# FIXME cfg = await get_config(cfg, config_type)
assert await cfg.option('a.a').value.get() == []
await cfg.option('a.a').value.set([1])
with pytest.raises(ValueError):
await cfg.option('a.b', 0).value.get()
await cfg.option('a.a').value.set([2])
await cfg.option('a.a').value.reset()
await cfg.option('a.a').value.set([2])
#
await cfg.property.add('demoting_error_warning')
with warnings.catch_warnings(record=True) as w:
await cfg.option('a.b', 0).value.set(2)
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -1051,20 +1090,21 @@ async def test_validator_default_diff():
a = IntOption('a', '', 3)
b = IntOption('b', '', 1, validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a, todict=True))))])
od = OptionDescription('od', '', [a, b])
cfg = await Config(od)
# FIXME cfg = await get_config(cfg, config_type)
await cfg.option('b').value.set(2)
await cfg.option('a').value.set(1)
owner = await cfg.owner.get()
assert await cfg.option('b').owner.get() == owner
with pytest.raises(ValueError):
await cfg.option('b').value.reset()
assert await cfg.option('b').owner.get() == owner
#
await cfg.property.add('demoting_error_warning')
with warnings.catch_warnings(record=True) as w:
await cfg.option('b').value.reset()
assert len(w) == 1
async with await Config(od) as cfg:
# FIXME cfg = await get_config(cfg, config_type)
await cfg.option('b').value.set(2)
await cfg.option('a').value.set(1)
owner = await cfg.owner.get()
assert await cfg.option('b').owner.get() == owner
with pytest.raises(ValueError):
await cfg.option('b').value.reset()
assert await cfg.option('b').owner.get() == owner
#
await cfg.property.add('demoting_error_warning')
with warnings.catch_warnings(record=True) as w:
await cfg.option('b').value.reset()
assert len(w) == 1
assert not await list_sessions()
@pytest.mark.asyncio
@ -1072,13 +1112,14 @@ async def test_validator_permissive(config_type):
a = IntOption('a', '', 1, properties=('hidden',))
b = IntOption('b', '', 2, validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a, todict=True))))])
od = OptionDescription('od', '', [a, b])
cfg = await Config(od)
await cfg.property.read_write()
await cfg.permissive.add('hidden')
cfg = await get_config(cfg, config_type)
with pytest.raises(ValueError):
await cfg.option('b').value.set(1)
await cfg.option('b').value.set(2)
async with await Config(od) as cfg:
await cfg.property.read_write()
await cfg.permissive.add('hidden')
cfg = await get_config(cfg, config_type)
with pytest.raises(ValueError):
await cfg.option('b').value.set(1)
await cfg.option('b').value.set(2)
assert not await list_sessions()
@pytest.mark.asyncio
@ -1086,11 +1127,12 @@ async def test_validator_disabled(config_type):
a = IntOption('a', '', 1, properties=('disabled',))
b = IntOption('b', '', 2, validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a, todict=True, raisepropertyerror=True))))])
od = OptionDescription('od', '', [a, b])
cfg = await Config(od)
await cfg.property.read_write()
cfg = await get_config(cfg, config_type)
with pytest.raises(PropertiesOptionError):
await cfg.option('b').value.set(1)
async with await Config(od) as cfg:
await cfg.property.read_write()
cfg = await get_config(cfg, config_type)
with pytest.raises(PropertiesOptionError):
await cfg.option('b').value.set(1)
assert not await list_sessions()
@pytest.mark.asyncio
@ -1098,10 +1140,11 @@ async def test_consistency_disabled_transitive(config_type):
a = IntOption('a', '', 1, properties=('disabled',))
b = IntOption('b', '', 2, validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a, todict=True, notraisepropertyerror=True))))])
od = OptionDescription('od', '', [a, b])
cfg = await Config(od)
await cfg.property.read_write()
cfg = await get_config(cfg, config_type)
await cfg.option('b').value.set(1)
async with await Config(od) as cfg:
await cfg.property.read_write()
cfg = await get_config(cfg, config_type)
await cfg.option('b').value.set(1)
assert not await list_sessions()
@pytest.mark.asyncio
@ -1112,29 +1155,30 @@ async def test_consistency_double_warnings(config_type):
od = OptionDescription('od', '', [a, b, c])
warnings.simplefilter("always", ValueWarning)
od2 = OptionDescription('od2', '', [od])
cfg_ori = await Config(od2)
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('od.c').value.set(1)
assert w != []
if config_type == 'tiramisu-api':
# in this case warnings is for '"a" and "b"'
async with await Config(od2) as cfg_ori:
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('od.c').value.set(1)
assert w != []
if config_type == 'tiramisu-api':
# in this case warnings is for '"a" and "b"'
assert len(w) == 1
else:
# in this cas one warnings is for "a" and the second for "b"
assert len(w) == 2
await cfg.option('od.a').value.set(2)
with warnings.catch_warnings(record=True) as w:
await cfg.option('od.c').value.get()
assert len(w) == 1
else:
# in this cas one warnings is for "a" and the second for "b"
assert len(w) == 2
await cfg.option('od.a').value.set(2)
with warnings.catch_warnings(record=True) as w:
await cfg.option('od.c').value.get()
assert len(w) == 1
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.pop('warnings')
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('od.c').value.set(1)
assert w == []
#
if config_type == 'tiramisu-api':
await cfg.send()
await cfg_ori.property.pop('warnings')
cfg = await get_config(cfg_ori, config_type)
with warnings.catch_warnings(record=True) as w:
await cfg.option('od.c').value.set(1)
assert w == []
assert not await list_sessions()
@pytest.mark.asyncio
@ -1147,12 +1191,13 @@ async def test_consistency_warnings_error(config_type):
])
od = OptionDescription('od', '', [a, b, c])
warnings.simplefilter("always", ValueWarning)
cfg = await Config(od)
cfg = await get_config(cfg, config_type)
with warnings.catch_warnings(record=True) as w:
with pytest.raises(ValueError):
await cfg.option('c').value.set(1)
assert w == []
async with await Config(od) as cfg:
cfg = await get_config(cfg, config_type)
with warnings.catch_warnings(record=True) as w:
with pytest.raises(ValueError):
await cfg.option('c').value.set(1)
assert w == []
assert not await list_sessions()
@pytest.mark.asyncio
@ -1161,8 +1206,9 @@ async def test_consistency_not_equal_has_dependency():
b = IntOption('b', '', )
b = IntOption('b', '', validators=[Calculation(valid_not_equal, Params((ParamSelfOption(), ParamOption(a, todict=True))))])
od = OptionDescription('od', '', [a, b])
cfg = await Config(od)
assert await cfg.option('a').option.has_dependency() is False
assert await cfg.option('b').option.has_dependency() is True
assert await cfg.option('a').option.has_dependency(False) is True
assert await cfg.option('b').option.has_dependency(False) is False
async with await Config(od) as cfg:
assert await cfg.option('a').option.has_dependency() is False
assert await cfg.option('b').option.has_dependency() is True
assert await cfg.option('a').option.has_dependency(False) is True
assert await cfg.option('b').option.has_dependency(False) is False
assert not await list_sessions()