2013-05-31 23:29:20 +02:00
# coding: utf-8
2017-07-09 09:49:03 +02:00
from . autopath import do_autopath
2015-07-24 17:54:10 +02:00
do_autopath ( )
2013-06-29 18:41:14 +02:00
from copy import copy
2020-01-22 20:46:18 +01:00
from os import environ
2018-04-10 22:42:20 +02:00
from tiramisu . i18n import _
2015-11-29 23:03:08 +01:00
from tiramisu . setting import groups
2013-05-31 23:29:20 +02:00
from tiramisu import setting
setting . expires_time = 1
2018-03-19 08:33:53 +01:00
from tiramisu import IPOption , OptionDescription , BoolOption , IntOption , StrOption , \
2019-09-01 09:41:53 +02:00
Leadership , Config , calc_value , Params , ParamOption , Calculation , ParamValue , ParamSelfOption , ParamIndex , \
calc_value_property_help
2019-10-27 11:09:15 +01:00
from tiramisu . error import PropertiesOptionError , ConfigError , display_list
2019-12-24 15:24:20 +01:00
import pytest
2018-10-31 08:00:19 +01:00
from tiramisu . storage import list_sessions , delete_session
2020-01-22 20:46:18 +01:00
from . config import config_type , get_config , event_loop
2013-05-31 23:29:20 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_properties ( config_type ) :
2018-04-05 19:06:38 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( ' disabled ' , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg_ori :
await cfg_ori . property . read_write ( )
cfg = await get_config ( cfg_ori , config_type )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
if config_type == ' tiramisu-api ' :
await cfg . send ( )
await cfg_ori . unrestraint . option ( ' ip_address_service ' ) . property . pop ( ' disabled ' )
cfg = await get_config ( cfg_ori , config_type )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
if config_type == ' tiramisu-api ' :
await cfg . send ( )
await cfg_ori . unrestraint . option ( ' ip_address_service ' ) . property . add ( ' disabled ' )
cfg = await get_config ( cfg_ori , config_type )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
# pop twice
if config_type == ' tiramisu-api ' :
await cfg . send ( )
await cfg_ori . unrestraint . option ( ' ip_address_service ' ) . property . pop ( ' disabled ' )
await cfg_ori . unrestraint . option ( ' ip_address_service ' ) . property . pop ( ' disabled ' )
assert not await list_sessions ( )
2018-04-05 19:06:38 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , todict = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = IPOption ( ' ip_address_service ' , ' ' ,
properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( False )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
await cfg . option ( ' activate_service ' ) . value . set ( True )
await cfg . option ( ' ip_address_service ' ) . value . get ( )
assert not await list_sessions ( )
2019-03-13 08:49:18 +01:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_inverse ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , todict = True ) ,
' expected ' : ParamValue ( False ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition ' : ParamValue ( True ) } ) )
2019-09-01 09:41:53 +02:00
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
await cfg . option ( ' activate_service ' ) . value . set ( False )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( True )
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
@pytest.mark.asyncio
async def test_requires_self ( config_type ) :
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamSelfOption ( ) ,
' expected ' : ParamValue ( ' b ' ) } ) )
a = StrOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a ] )
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
assert await cfg . option ( ' ip_address_service ' ) . value . get ( ) == None
await cfg . option ( ' ip_address_service ' ) . value . set ( ' a ' )
assert await cfg . option ( ' ip_address_service ' ) . value . get ( ) == ' a '
await cfg . option ( ' ip_address_service ' ) . value . set ( ' b ' )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
@pytest.mark.asyncio
async def test_requires_with_requires ( config_type ) :
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a ) ,
' expected ' : ParamValue ( False ) } ) )
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
if environ . get ( ' TIRAMISU_STORAGE ' ) :
print ( ' not implemeted yet (store calculated property) ' )
else :
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
await cfg . option ( ' ip_address_service ' ) . property . add ( ' test ' )
cfg = await get_config ( cfg , config_type )
await cfg . option ( ' ip_address_service ' ) . value . get ( )
await cfg . option ( ' activate_service ' ) . value . set ( False )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
await cfg . option ( ' activate_service ' ) . value . set ( True )
await cfg . option ( ' ip_address_service ' ) . value . get ( )
assert not await list_sessions ( )
@pytest.mark.asyncio
async def test_requires_same_action ( config_type ) :
activate_service = BoolOption ( ' activate_service ' , ' ' , True )
new_property = Calculation ( calc_value ,
Params ( ParamValue ( ' new ' ) ,
kwargs = { ' condition ' : ParamOption ( activate_service , todict = True ) ,
' expected ' : ParamValue ( False ) } ) ,
calc_value_property_help )
activate_service_web = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( new_property , ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( activate_service_web , notraisepropertyerror = True , todict = True ) ,
' expected ' : ParamValue ( False ) } ) ,
calc_value_property_help )
ip_address_service_web = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( disabled_property , ) )
od1 = OptionDescription ( ' service ' , ' ' , [ activate_service , activate_service_web , ip_address_service_web ] )
async with await Config ( od1 ) as cfg :
await cfg . property . read_write ( )
await cfg . property . add ( ' new ' )
cfg = await get_config ( cfg , config_type )
await cfg . option ( ' activate_service ' ) . value . get ( )
await cfg . option ( ' activate_service_web ' ) . value . get ( )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
await cfg . option ( ' activate_service ' ) . value . set ( False )
#
props = [ ]
try :
await cfg . option ( ' activate_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
if config_type == ' tiramisu ' :
assert frozenset ( props ) == frozenset ( [ ' new ' ] )
else :
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
#
props = [ ]
try :
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
submsg = ' " disabled " ( ' + _ ( ' the value of " {0} " is {1} ' ) . format ( ' activate_service ' , ' " False " ' ) + ' ) '
if config_type == ' tiramisu ' :
submsg = ' " new " ( ' + _ ( ' the value of " {0} " is {1} ' ) . format ( ' activate_service ' , ' " False " ' ) + ' ) '
submsg = ' " disabled " ( ' + str ( _ ( ' cannot access to {0} " {1} " because has {2} {3} ' ) . format ( ' option ' , ' activate_service_web ' , _ ( ' property ' ) , submsg ) ) + ' ) '
assert str ( err ) == str ( _ ( ' cannot access to {0} " {1} " because has {2} {3} ' ) . format ( ' option ' , ' ip_address_service_web ' , _ ( ' property ' ) , submsg ) )
#access to cache
assert str ( err ) == str ( _ ( ' cannot access to {0} " {1} " because has {2} {3} ' ) . format ( ' option ' , ' ip_address_service_web ' , _ ( ' property ' ) , submsg ) )
else :
# FIXME
assert str ( err ) == ' error '
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_multiple_requires ( config_type ) :
2019-09-01 09:41:53 +02:00
a = StrOption ( ' activate_service ' , ' ' )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a ) ,
' expected_0 ' : ParamValue ( ' yes ' ) ,
' expected_1 ' : ParamValue ( ' ok ' ) } ) )
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' yes ' )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' ok ' )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' no ' )
await cfg . option ( ' ip_address_service ' ) . value . get ( )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_multiple_requires_cumulative ( config_type ) :
2019-09-01 09:41:53 +02:00
a = StrOption ( ' activate_service ' , ' ' )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a ) ,
' expected ' : ParamValue ( ' yes ' ) } ) )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( a ) ,
' expected ' : ParamValue ( ' yes ' ) } ) )
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , hidden_property ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' yes ' )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
if config_type == ' tiramisu ' :
assert set ( props ) == { ' hidden ' , ' disabled ' }
else :
assert set ( props ) == { ' disabled ' }
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' ok ' )
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' no ' )
await cfg . option ( ' ip_address_service ' ) . value . get ( )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_multiple_requires_cumulative_inverse ( config_type ) :
2013-09-02 19:47:00 +02:00
a = StrOption ( ' activate_service ' , ' ' )
2019-10-27 11:09:15 +01:00
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a ) ,
' expected ' : ParamValue ( ' yes ' ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition ' : ParamValue ( True ) } ) )
2019-10-27 11:09:15 +01:00
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( a ) ,
' expected ' : ParamValue ( ' yes ' ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition ' : ParamValue ( True ) } ) )
2019-10-27 11:09:15 +01:00
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , hidden_property ) )
2013-09-02 19:47:00 +02:00
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
if config_type == ' tiramisu ' :
assert set ( props ) == { ' hidden ' , ' disabled ' }
else :
assert set ( props ) == { ' disabled ' }
await cfg . option ( ' activate_service ' ) . value . set ( ' yes ' )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2019-07-04 20:43:47 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' ok ' )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
if config_type == ' tiramisu ' :
assert set ( props ) == { ' hidden ' , ' disabled ' }
else :
assert set ( props ) == { ' disabled ' }
2013-09-02 19:47:00 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' no ' )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
if config_type == ' tiramisu ' :
assert set ( props ) == { ' hidden ' , ' disabled ' }
else :
assert set ( props ) == { ' disabled ' }
assert not await list_sessions ( )
2013-09-02 19:47:00 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_multiple_requires_inverse ( config_type ) :
2019-09-01 09:41:53 +02:00
a = StrOption ( ' activate_service ' , ' ' )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a ) ,
2019-10-27 11:09:15 +01:00
' expected_0 ' : ParamValue ( ' yes ' ) ,
' expected_1 ' : ParamValue ( ' ok ' ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition ' : ParamValue ( True ) } ) )
2019-10-27 11:09:15 +01:00
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
2019-09-01 09:41:53 +02:00
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' yes ' )
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2019-10-27 11:09:15 +01:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' ok ' )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( ' no ' )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
2013-05-31 23:29:20 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_transitive ( config_type ) :
2019-03-13 08:49:18 +01:00
a = BoolOption ( ' activate_service ' , ' ' , True )
2019-09-01 09:41:53 +02:00
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( disabled_property , ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( b , raisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( disabled_property , ) )
2019-03-13 08:49:18 +01:00
od = OptionDescription ( ' service ' , ' ' , [ a , b , d ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
await cfg . option ( ' activate_service ' ) . value . get ( )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' activate_service_web ' ) . value . get ( )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( False )
#
props = [ ]
try :
await cfg . option ( ' activate_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
#
props = [ ]
try :
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
2019-03-13 08:49:18 +01:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_transitive_unrestraint ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( disabled_property , ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( b , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b , d ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg_ori :
await cfg_ori . property . read_write ( )
cfg = await get_config ( cfg_ori , config_type )
await cfg . option ( ' activate_service ' ) . value . get ( )
await cfg . option ( ' activate_service_web ' ) . value . get ( )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
await cfg . option ( ' activate_service ' ) . value . set ( False )
#
if config_type == ' tiramisu-api ' :
await cfg . send ( )
assert await cfg_ori . unrestraint . option ( ' activate_service_web ' ) . property . get ( ) == { ' disabled ' }
assert await cfg_ori . unrestraint . option ( ' ip_address_service_web ' ) . property . get ( ) == { ' disabled ' }
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_transitive_owner ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( disabled_property , ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( b , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b , d ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
await cfg . option ( ' activate_service ' ) . value . get ( )
await cfg . option ( ' activate_service_web ' ) . value . get ( )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
#no more default value
await cfg . option ( ' ip_address_service_web ' ) . value . set ( ' 1.1.1.1 ' )
await cfg . option ( ' activate_service ' ) . value . set ( False )
props = [ ]
try :
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_transitive_bis ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
abis = BoolOption ( ' activate_service_bis ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( True ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition ' : ParamValue ( True ) } ) )
2019-09-01 09:41:53 +02:00
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( disabled_property , ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( b , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( True ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition ' : ParamValue ( True ) } ) )
2019-09-01 09:41:53 +02:00
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , abis , b , d ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
#
2019-12-24 15:24:20 +01:00
await cfg . option ( ' activate_service_web ' ) . value . get ( )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( False )
#
props = [ ]
try :
await cfg . option ( ' activate_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
#
props = [ ]
try :
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_transitive_hidden_permissive ( ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( hidden_property , ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( b , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b , d ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
# FIXME permissive cfg = await get_config(cfg, config_type)
await cfg . option ( ' activate_service ' ) . value . get ( )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
await cfg . option ( ' activate_service ' ) . value . set ( False )
#
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_transitive_hidden_disabled ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( hidden_property , ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( b , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b , d ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
await cfg . option ( ' activate_service ' ) . value . get ( )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' activate_service_web ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
await cfg . option ( ' activate_service ' ) . value . set ( False )
#
props = [ ]
try :
await cfg . option ( ' activate_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
if config_type == ' tiramisu-api ' :
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
else :
assert frozenset ( props ) == frozenset ( [ ' hidden ' ] )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
assert not await list_sessions ( )
@pytest.mark.asyncio
async def test_requires_transitive_hidden_disabled_multiple ( config_type ) :
a = BoolOption ( ' activate_service ' , ' ' , True )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( hidden_property , disabled_property ) )
mandatory_property = Calculation ( calc_value ,
Params ( ParamValue ( ' mandatory ' ) ,
kwargs = { ' condition ' : ParamOption ( b ) ,
' expected ' : ParamValue ( False ) } ) )
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( mandatory_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b , d ] )
async with await Config ( od ) as cfg_ori :
await cfg_ori . property . read_write ( )
cfg = await get_config ( cfg_ori , config_type )
await cfg . option ( ' activate_service ' ) . value . get ( )
await cfg . option ( ' activate_service_web ' ) . value . get ( )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
req = None
if config_type == ' tiramisu-api ' :
try :
await cfg . option ( ' activate_service ' ) . value . set ( False )
except ConfigError as err :
req = err
error_msg = str ( _ ( ' unable to transform tiramisu object to dict: {} ' ) . format ( _ ( ' cannot access to option " {0} " because required option " {1} " has {2} {3} ' ) . format ( ' ip_address_service_web ' , ' activate_service_web ' , _ ( ' property ' ) , ' " disabled " ' ) ) )
else :
await cfg . option ( ' activate_service ' ) . value . set ( False )
#
props = [ ]
try :
await cfg . option ( ' activate_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert set ( props ) == { ' disabled ' , ' hidden ' }
del props
#
try :
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
except ConfigError as err :
req = err
error_msg = str ( _ ( ' unable to carry out a calculation for " {} " , {} ' ) . format ( ' ip_address_service_web ' , _ ( ' cannot access to {0} " {1} " because has {2} {3} ' ) . format ( ' option ' , ' activate_service_web ' , _ ( ' property ' ) , display_list ( [ ' disabled ' ] , add_quote = True ) ) ) )
assert req , " ip_address_service_web should raise ConfigError "
assert str ( req ) == error_msg
del req
#
await cfg_ori . permissive . reset ( )
if config_type == ' tiramisu-api ' :
try :
cfg = await get_config ( cfg_ori , config_type )
except ConfigError as err :
req = err
error_msg = str ( _ ( ' unable to transform tiramisu object to dict: {} ' ) . format ( _ ( ' cannot access to option " {0} " because required option " {1} " has {2} {3} ' ) . format ( ' ip_address_service_web ' , ' activate_service_web ' , _ ( ' properties ' ) , ' " disabled " {} " hidden " ' . format ( _ ( ' and ' ) ) ) ) )
else :
cfg = await get_config ( cfg_ori , config_type )
try :
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
except ConfigError as err :
req = err
error_msg = str ( _ ( ' unable to carry out a calculation for " {} " , {} ' ) . format ( ' ip_address_service_web ' , _ ( ' cannot access to {0} " {1} " because has {2} {3} ' ) . format ( ' option ' , ' activate_service_web ' , _ ( ' properties ' ) , display_list ( [ ' hidden ' , ' disabled ' ] , add_quote = True ) ) ) )
assert req , " ip_address_service_web should raise ConfigError "
assert str ( req ) == error_msg
del req
assert not await list_sessions ( )
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_not_transitive ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( disabled_property , ) )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( b , notraisepropertyerror = True ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' expected ' : ParamValue ( False ) } ) )
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b , d ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
await cfg . option ( ' activate_service ' ) . value . get ( )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' activate_service_web ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
await cfg . option ( ' activate_service ' ) . value . set ( False )
#
props = [ ]
try :
await cfg . option ( ' activate_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
#
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_not_transitive_not_same_action ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = BoolOption ( ' activate_service_web ' , ' ' , True , properties = ( disabled_property , ) )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( b ) ,
' expected ' : ParamValue ( False ) } ) )
d = IPOption ( ' ip_address_service_web ' , ' ' , properties = ( hidden_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b , d ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
await cfg . option ( ' activate_service ' ) . value . get ( )
await cfg . option ( ' activate_service_web ' ) . value . get ( )
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
if config_type == ' tiramisu-api ' :
with pytest . raises ( ConfigError ) :
await cfg . option ( ' activate_service ' ) . value . set ( False )
else :
2019-12-24 15:24:20 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( False )
2020-01-22 20:46:18 +01:00
#
props = [ ]
try :
await cfg . option ( ' activate_service_web ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
#
with pytest . raises ( ConfigError ) :
await cfg . option ( ' ip_address_service_web ' ) . value . get ( )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_none ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( None ) } ) )
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
await cfg . option ( ' activate_service ' ) . value . set ( False )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_multi_disabled ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' )
b = IntOption ( ' num_service ' , ' ' )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition_0 ' : ParamOption ( a , notraisepropertyerror = True ) ,
' condition_1 ' : ParamOption ( b , notraisepropertyerror = True ) ,
' expected_0 ' : ParamValue ( True ) ,
' expected_1 ' : ParamValue ( 1 ) ,
' condition_operator ' : ParamValue ( ' OR ' ) } ) )
c = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b , c ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( True )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( False )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' ip_address_service ' ) . value . get ( )
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' num_service ' ) . value . set ( 1 )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
await cfg . option ( ' activate_service ' ) . value . set ( True )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_multi_disabled_inverse ( config_type ) :
2019-03-13 08:49:18 +01:00
a = BoolOption ( ' activate_service ' , ' ' )
b = IntOption ( ' num_service ' , ' ' )
2019-10-27 11:09:15 +01:00
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition_0 ' : ParamOption ( a , notraisepropertyerror = True ) ,
' condition_1 ' : ParamOption ( b , notraisepropertyerror = True ) ,
' expected_0 ' : ParamValue ( True ) ,
' expected_1 ' : ParamValue ( 1 ) ,
' condition_operator ' : ParamValue ( ' OR ' ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition_0 ' : ParamValue ( True ) ,
' reverse_condition_1 ' : ParamValue ( True ) } ) )
2019-10-27 11:09:15 +01:00
c = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
2019-03-13 08:49:18 +01:00
od = OptionDescription ( ' service ' , ' ' , [ a , b , c ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
2019-03-13 08:49:18 +01:00
2020-01-22 20:46:18 +01:00
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2019-03-13 08:49:18 +01:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( True )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2019-03-13 08:49:18 +01:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( False )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2017-05-20 16:28:19 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' num_service ' ) . value . set ( 1 )
props = [ ]
try :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2017-05-20 16:28:19 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' activate_service ' ) . value . set ( True )
await cfg . option ( ' ip_address_service ' ) . value . get ( )
assert not await list_sessions ( )
2017-07-21 22:34:41 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_multi_disabled_2 ( config_type ) :
2019-10-27 11:09:15 +01:00
a = BoolOption ( ' a ' , ' ' )
b = BoolOption ( ' b ' , ' ' )
c = BoolOption ( ' c ' , ' ' )
d = BoolOption ( ' d ' , ' ' )
e = BoolOption ( ' e ' , ' ' )
f = BoolOption ( ' f ' , ' ' )
g = BoolOption ( ' g ' , ' ' )
h = BoolOption ( ' h ' , ' ' )
i = BoolOption ( ' i ' , ' ' )
j = BoolOption ( ' j ' , ' ' )
k = BoolOption ( ' k ' , ' ' )
l = BoolOption ( ' l ' , ' ' )
m = BoolOption ( ' m ' , ' ' )
list_bools = [ a , b , c , d , e , f , g , h , i , j , k , l , m ]
requires = [ ]
kwargs = { ' expected ' : ParamValue ( True ) ,
' condition_operator ' : ParamValue ( ' OR ' ) }
for idx , boo in enumerate ( list_bools ) :
kwargs [ ' condition_ {} ' . format ( idx ) ] = ParamOption ( boo , notraisepropertyerror = True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = kwargs ) )
z = IPOption ( ' z ' , ' ' , properties = ( disabled_property , ) )
y = copy ( list_bools )
y . append ( z )
od = OptionDescription ( ' service ' , ' ' , y )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
2019-09-01 09:41:53 +02:00
2020-01-22 20:46:18 +01:00
await cfg . option ( ' z ' ) . value . get ( )
for boo in list_bools :
await cfg . option ( boo . impl_getname ( ) ) . value . set ( True )
2019-09-01 09:41:53 +02:00
props = [ ]
try :
2019-12-24 15:24:20 +01:00
await cfg . option ( ' z ' ) . value . get ( )
2019-09-01 09:41:53 +02:00
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2020-01-22 20:46:18 +01:00
for boo in list_bools :
await cfg . option ( boo . impl_getname ( ) ) . value . set ( False )
if boo == m :
await cfg . option ( ' z ' ) . value . get ( )
else :
props = [ ]
try :
await cfg . option ( ' z ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_multi_disabled_inverse_2 ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' a ' , ' ' )
b = BoolOption ( ' b ' , ' ' )
c = BoolOption ( ' c ' , ' ' )
d = BoolOption ( ' d ' , ' ' )
e = BoolOption ( ' e ' , ' ' )
f = BoolOption ( ' f ' , ' ' )
g = BoolOption ( ' g ' , ' ' )
h = BoolOption ( ' h ' , ' ' )
i = BoolOption ( ' i ' , ' ' )
j = BoolOption ( ' j ' , ' ' )
k = BoolOption ( ' k ' , ' ' )
l = BoolOption ( ' l ' , ' ' )
m = BoolOption ( ' m ' , ' ' )
list_bools = [ a , b , c , d , e , f , g , h , i , j , k , l , m ]
requires = [ ]
kwargs = { ' expected ' : ParamValue ( True ) ,
' condition_operator ' : ParamValue ( ' OR ' ) }
for idx , boo in enumerate ( list_bools ) :
kwargs [ ' condition_ {} ' . format ( idx ) ] = ParamOption ( boo , notraisepropertyerror = True )
2019-12-08 09:09:48 +01:00
kwargs [ ' reverse_condition_ {} ' . format ( idx ) ] = ParamValue ( True )
2019-09-01 09:41:53 +02:00
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = kwargs ) )
#for boo in list_bools:
# requires.append({'option': boo, 'expected': True, 'action': 'disabled',
# 'inverse': True})
z = IPOption ( ' z ' , ' ' , properties = ( disabled_property , ) )
y = copy ( list_bools )
y . append ( z )
od = OptionDescription ( ' service ' , ' ' , y )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
2019-09-01 09:41:53 +02:00
props = [ ]
try :
2019-12-24 15:24:20 +01:00
await cfg . option ( ' z ' ) . value . get ( )
2019-09-01 09:41:53 +02:00
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2020-01-22 20:46:18 +01:00
for boo in list_bools :
await cfg . option ( boo . impl_getname ( ) ) . value . set ( True )
if boo != m :
# it's disabled until last option is modified
props = [ ]
try :
await cfg . option ( ' z ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2019-12-24 15:24:20 +01:00
await cfg . option ( ' z ' ) . value . get ( )
2020-01-22 20:46:18 +01:00
for boo in list_bools :
await cfg . option ( boo . impl_getname ( ) ) . value . set ( False )
2019-09-01 09:41:53 +02:00
props = [ ]
try :
2019-12-24 15:24:20 +01:00
await cfg . option ( ' z ' ) . value . get ( )
2019-09-01 09:41:53 +02:00
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
2020-01-22 20:46:18 +01:00
try :
await cfg . option ( ' z ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
for boo in reversed ( list_bools ) :
await cfg . option ( boo . impl_getname ( ) ) . value . set ( True )
if boo != a :
# it's disabled until last option is modified
props = [ ]
try :
await cfg . option ( ' z ' ) . value . get ( )
except PropertiesOptionError as err :
props = err . proptype
assert frozenset ( props ) == frozenset ( [ ' disabled ' ] )
await cfg . option ( ' z ' ) . value . get ( )
assert not await list_sessions ( )
@pytest.mark.asyncio
async def test_requires_requirement_append ( config_type ) :
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
async with await Config ( od ) as cfg_ori :
await cfg_ori . property . read_write ( )
cfg = await get_config ( cfg_ori , config_type )
await cfg . property . get ( )
await cfg . option ( ' ip_address_service ' ) . property . get ( )
if config_type == ' tiramisu-api ' :
await cfg . send ( )
#raises(ValueError, "await cfg_ori.option('ip_address_service').property.add('disabled')")
cfg = await get_config ( cfg_ori , config_type )
await cfg . option ( ' activate_service ' ) . value . set ( False )
# disabled is now set, test to remove disabled before store in storage
if config_type == ' tiramisu-api ' :
await cfg . send ( )
if environ . get ( ' TIRAMISU_STORAGE ' ) :
print ( ' not implemeted yet (store calculated property) ' )
else :
await cfg_ori . unrestraint . option ( ' ip_address_service ' ) . property . add ( " test " )
assert not await list_sessions ( )
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_different_inverse ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition_0 ' : ParamOption ( a , notraisepropertyerror = True ) ,
' condition_1 ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected_0 ' : ParamValue ( True ) ,
' expected_1 ' : ParamValue ( True ) ,
' condition_operator ' : ParamValue ( ' OR ' ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition_0 ' : ParamValue ( True ) } ) )
2019-09-01 09:41:53 +02:00
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , b ] )
2020-01-22 20:46:18 +01:00
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('ip_address_service').value.get()
await cfg . option ( ' activate_service ' ) . value . set ( False )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
assert not await list_sessions ( )
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_requires_different_inverse_unicode ( config_type ) :
2019-09-01 09:41:53 +02:00
a = BoolOption ( ' activate_service ' , ' ' , True )
d = StrOption ( ' activate_other_service ' , ' ' , ' val2 ' )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition_0 ' : ParamOption ( a , notraisepropertyerror = True ) ,
' condition_1 ' : ParamOption ( d , notraisepropertyerror = True ) ,
' expected_0 ' : ParamValue ( True ) ,
' expected_1 ' : ParamValue ( ' val1 ' ) ,
' condition_operator ' : ParamValue ( ' OR ' ) ,
2019-12-08 09:09:48 +01:00
' reverse_condition_0 ' : ParamValue ( True ) } ) )
2019-09-01 09:41:53 +02:00
b = IPOption ( ' ip_address_service ' , ' ' , properties = ( disabled_property , ) )
od = OptionDescription ( ' service ' , ' ' , [ a , d , b ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
assert await cfg . option ( ' ip_address_service ' ) . value . get ( ) == None
await cfg . option ( ' activate_service ' ) . value . set ( False )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
await cfg . option ( ' activate_service ' ) . value . set ( True )
assert await cfg . option ( ' ip_address_service ' ) . value . get ( ) == None
await cfg . option ( ' activate_other_service ' ) . value . set ( ' val1 ' )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
await cfg . option ( ' activate_service ' ) . value . set ( False )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_address_service ' ) . value . get ( )
assert not await list_sessions ( )
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_optiondescription_requires ( ) :
2019-10-27 11:09:15 +01:00
a = BoolOption ( ' activate_service ' , ' ' , True )
b = BoolOption ( ' ip_address_service ' , ' ' , multi = True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( a , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) } ) )
OptionDescription ( ' service ' , ' ' , [ b ] , properties = ( disabled_property , ) )
2015-11-30 15:55:34 +01:00
2019-10-27 11:09:15 +01:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_leadership_requires ( config_type ) :
2019-12-08 09:09:48 +01:00
ip_admin_eth0 = StrOption ( ' ip_admin_eth0 ' , " ip réseau autorisé " , multi = True , properties = ( ' notunique ' , ) )
2019-10-27 11:09:15 +01:00
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( ip_admin_eth0 , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( ' 192.168.1.1 ' ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' index ' : ParamIndex ( ) } ) )
netmask_admin_eth0 = StrOption ( ' netmask_admin_eth0 ' , " masque du sous-réseau " , multi = True , properties = ( disabled_property , ) )
2019-03-13 08:49:18 +01:00
interface1 = Leadership ( ' ip_admin_eth0 ' , ' ' , [ ip_admin_eth0 , netmask_admin_eth0 ] )
2019-07-04 20:43:47 +02:00
od = OptionDescription ( ' toto ' , ' ' , [ interface1 ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
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 ( [ ' 192.168.1.2 ' ] )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
assert await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( ) == [ ' 192.168.1.2 ' ]
#
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . set ( [ ' 192.168.1.2 ' , ' 192.168.1.1 ' ] )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . get ( )
#
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . set ( [ ' 192.168.1.2 ' , ' 192.168.1.2 ' ] )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . get ( ) is None
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . set ( ' 255.255.255.255 ' )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . get ( ) == ' 255.255.255.255 '
assert await cfg . value . dict ( ) == { ' ip_admin_eth0.ip_admin_eth0 ' : [ ' 192.168.1.2 ' , ' 192.168.1.2 ' ] ,
' ip_admin_eth0.netmask_admin_eth0 ' : [ None , ' 255.255.255.255 ' ] }
assert await cfg . value . dict ( leader_to_list = True ) == { ' ip_admin_eth0.ip_admin_eth0 ' : [ { ' ip_admin_eth0.ip_admin_eth0 ' : ' 192.168.1.2 ' , ' ip_admin_eth0.netmask_admin_eth0 ' : None } ,
{ ' ip_admin_eth0.ip_admin_eth0 ' : ' 192.168.1.2 ' , ' ip_admin_eth0.netmask_admin_eth0 ' : ' 255.255.255.255 ' } ] }
#
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . set ( [ ' 192.168.1.2 ' , ' 192.168.1.1 ' ] )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . get ( )
ret = await cfg . value . dict ( )
assert set ( ret . keys ( ) ) == set ( [ ' ip_admin_eth0.ip_admin_eth0 ' , ' ip_admin_eth0.netmask_admin_eth0 ' ] )
assert ret [ ' ip_admin_eth0.ip_admin_eth0 ' ] == [ ' 192.168.1.2 ' , ' 192.168.1.1 ' ]
assert len ( ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] ) == 2
assert ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] [ 0 ] is None
assert isinstance ( ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] [ 1 ] , PropertiesOptionError )
del ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] [ 1 ]
del ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] [ 0 ]
del ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ]
assert await cfg . value . dict ( leader_to_list = True ) == { ' ip_admin_eth0.ip_admin_eth0 ' : [ { ' ip_admin_eth0.ip_admin_eth0 ' : ' 192.168.1.2 ' ,
' ip_admin_eth0.netmask_admin_eth0 ' : None } ,
{ ' ip_admin_eth0.ip_admin_eth0 ' : ' 192.168.1.1 ' } ] }
#
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . set ( ' 255.255.255.255 ' )
ret = await cfg . value . dict ( )
assert set ( ret . keys ( ) ) == set ( [ ' ip_admin_eth0.ip_admin_eth0 ' , ' ip_admin_eth0.netmask_admin_eth0 ' ] )
assert ret [ ' ip_admin_eth0.ip_admin_eth0 ' ] == [ ' 192.168.1.2 ' , ' 192.168.1.1 ' ]
assert len ( ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] ) == 2
assert ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] [ 0 ] == ' 255.255.255.255 '
assert isinstance ( ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] [ 1 ] , PropertiesOptionError )
del ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] [ 1 ]
del ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ] [ 0 ]
del ret [ ' ip_admin_eth0.netmask_admin_eth0 ' ]
assert await cfg . value . dict ( leader_to_list = True ) == { ' ip_admin_eth0.ip_admin_eth0 ' : [ { ' ip_admin_eth0.ip_admin_eth0 ' : ' 192.168.1.2 ' ,
' ip_admin_eth0.netmask_admin_eth0 ' : ' 255.255.255.255 ' } ,
{ ' ip_admin_eth0.ip_admin_eth0 ' : ' 192.168.1.1 ' } ] }
assert not await list_sessions ( )
2019-03-13 08:49:18 +01:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_leadership_requires_leader ( config_type ) :
2019-09-01 09:41:53 +02:00
activate = BoolOption ( ' activate ' , " Activer l ' accès au réseau " , True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( activate , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) ,
' index ' : ParamIndex ( ) } ) )
2019-10-27 11:09:15 +01:00
ip_admin_eth0 = StrOption ( ' ip_admin_eth0 ' , " ip réseau autorisé " , multi = True )
2019-09-01 09:41:53 +02:00
netmask_admin_eth0 = StrOption ( ' netmask_admin_eth0 ' , " masque du sous-réseau " , multi = True )
2019-10-27 11:09:15 +01:00
interface1 = Leadership ( ' ip_admin_eth0 ' , ' ' , [ ip_admin_eth0 , netmask_admin_eth0 ] , properties = ( disabled_property , ) )
2019-09-01 09:41:53 +02:00
od = OptionDescription ( ' toto ' , ' ' , [ activate , interface1 ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
#
assert await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( ) == [ ]
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . set ( [ ' 192.168.1.2 ' ] )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
assert await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( ) == [ ' 192.168.1.2 ' ]
#
await cfg . option ( ' activate ' ) . value . set ( False )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( )
#
await cfg . option ( ' activate ' ) . value . set ( True )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
#
await cfg . option ( ' activate ' ) . value . set ( False )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( )
assert await cfg . value . dict ( ) == { ' activate ' : False }
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_leadership_requires_leadership ( config_type ) :
2019-09-01 09:41:53 +02:00
activate = BoolOption ( ' activate ' , " Activer l ' accès au réseau " , True )
ip_admin_eth0 = StrOption ( ' ip_admin_eth0 ' , " ip réseau autorisé " , multi = True )
netmask_admin_eth0 = StrOption ( ' netmask_admin_eth0 ' , " masque du sous-réseau " , multi = True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( activate , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( False ) ,
' index ' : ParamIndex ( ) } ) )
interface1 = Leadership ( ' ip_admin_eth0 ' , ' ' , [ ip_admin_eth0 , netmask_admin_eth0 ] , properties = ( disabled_property , ) )
od = OptionDescription ( ' toto ' , ' ' , [ activate , interface1 ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
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 ( [ ' 192.168.1.2 ' ] )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
assert await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( ) == [ ' 192.168.1.2 ' ]
#
await cfg . option ( ' activate ' ) . value . set ( False )
if config_type != ' tiramisu-api ' :
# FIXME
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( )
#
await cfg . option ( ' activate ' ) . value . set ( True )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
#
await cfg . option ( ' activate ' ) . value . set ( False )
if config_type != ' tiramisu-api ' :
# FIXME
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( )
assert await cfg . value . dict ( ) == { ' activate ' : False }
assert not await list_sessions ( )
2019-09-01 09:41:53 +02:00
2019-12-24 15:24:20 +01:00
@pytest.mark.asyncio
async def test_leadership_requires_no_leader ( config_type ) :
2019-09-01 09:41:53 +02:00
activate = BoolOption ( ' activate ' , " Activer l ' accès au réseau " , True )
ip_admin_eth0 = StrOption ( ' ip_admin_eth0 ' , " ip réseau autorisé " , multi = True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( activate , notraisepropertyerror = True ) ,
2019-12-08 09:09:48 +01:00
' expected ' : ParamValue ( False ) } ) )
2019-09-01 09:41:53 +02:00
netmask_admin_eth0 = StrOption ( ' netmask_admin_eth0 ' , " masque du sous-réseau " , multi = True , properties = ( disabled_property , ) )
interface1 = Leadership ( ' ip_admin_eth0 ' , ' ' , [ ip_admin_eth0 , netmask_admin_eth0 ] )
od = OptionDescription ( ' toto ' , ' ' , [ activate , interface1 ] )
2020-01-22 20:46:18 +01:00
async with await Config ( od ) as cfg :
await cfg . property . read_write ( )
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 ( [ ' 192.168.1.2 ' ] )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
assert await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( ) == [ ' 192.168.1.2 ' ]
await cfg . option ( ' activate ' ) . value . set ( False )
await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . set ( [ ' 192.168.1.2 ' , ' 192.168.1.1 ' ] )
assert await cfg . option ( ' ip_admin_eth0.ip_admin_eth0 ' ) . value . get ( ) == [ ' 192.168.1.2 ' , ' 192.168.1.1 ' ]
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . get ( )
await cfg . option ( ' activate ' ) . value . set ( True )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( ) is None
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . get ( ) is None
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . set ( ' 255.255.255.255 ' )
assert await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . get ( ) == ' 255.255.255.255 '
await cfg . option ( ' activate ' ) . value . set ( False )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 0 ) . value . get ( )
with pytest . raises ( PropertiesOptionError ) :
await cfg . option ( ' ip_admin_eth0.netmask_admin_eth0 ' , 1 ) . value . get ( )
assert await cfg . value . dict ( ) == { ' ip_admin_eth0.ip_admin_eth0 ' : [ ' 192.168.1.2 ' , ' 192.168.1.1 ' ] , ' activate ' : False }
assert not await list_sessions ( )
@pytest.mark.asyncio
async def test_leadership_requires_complet ( config_type ) :
optiontoto = StrOption ( ' unicodetoto ' , " Unicode " )
option = StrOption ( ' unicode ' , " Unicode leader " , multi = True )
option1 = StrOption ( ' unicode1 ' , " Unicode follower 1 " , multi = True )
option2 = StrOption ( ' unicode2 ' , " Values ' test ' must show ' Unicode follower 3 ' " , multi = True )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( option , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( ' test ' ) ,
' index ' : ParamIndex ( ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' reverse_condition ' : ParamValue ( True ) } ) )
option3 = StrOption ( ' unicode3 ' , " Unicode follower 3 " , properties = ( hidden_property , ) , multi = True )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( option2 , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( ' test ' ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' index ' : ParamIndex ( ) ,
' reverse_condition ' : ParamValue ( True ) } ) )
option4 = StrOption ( ' unicode4 ' , " Unicode follower 4 " , properties = ( hidden_property , ) , multi = True )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition ' : ParamOption ( optiontoto , notraisepropertyerror = True ) ,
' expected ' : ParamValue ( ' test ' ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' reverse_condition ' : ParamValue ( True ) } ) )
option5 = StrOption ( ' unicode5 ' , " Unicode follower 5 " , properties = ( hidden_property , ) , multi = True )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition_0 ' : ParamOption ( optiontoto , notraisepropertyerror = True ) ,
' expected_0 ' : ParamValue ( ' test ' ) ,
' condition_1 ' : ParamOption ( option2 , notraisepropertyerror = True ) ,
' expected_1 ' : ParamValue ( ' test ' ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' condition_operator ' : ParamValue ( ' OR ' ) ,
' reverse_condition_0 ' : ParamValue ( True ) ,
' reverse_condition_1 ' : ParamValue ( True ) } ) )
option6 = StrOption ( ' unicode6 ' , " Unicode follower 6 " , properties = ( hidden_property , ) , multi = True )
hidden_property = Calculation ( calc_value ,
Params ( ParamValue ( ' hidden ' ) ,
kwargs = { ' condition_0 ' : ParamOption ( option2 , notraisepropertyerror = True ) ,
' expected_0 ' : ParamValue ( ' test ' ) ,
' condition_1 ' : ParamOption ( optiontoto , notraisepropertyerror = True ) ,
' expected_1 ' : ParamValue ( ' test ' ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' reverse_condition ' : ParamValue ( True ) } ) )
option7 = StrOption ( ' unicode7 ' , " Unicode follower 7 " , properties = ( hidden_property , ) , multi = True )
descr1 = Leadership ( " unicode " , " Common configuration 1 " ,
[ option , option1 , option2 , option3 , option4 , option5 , option6 , option7 ] )
descr = OptionDescription ( " options " , " Common configuration 2 " , [ descr1 , optiontoto ] )
descr = OptionDescription ( " unicode1_leadership_requires " , " Leader followers with Unicode follower 3 hidden when Unicode follower 2 is test " , [ descr ] )
async with await Config ( descr ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
await cfg . option ( ' options.unicode.unicode ' ) . value . set ( [ ' test ' , ' trah ' ] )
await cfg . option ( ' options.unicode.unicode2 ' , 0 ) . value . set ( ' test ' )
dico = await cfg . value . dict ( )
assert dico . keys ( ) == set ( [ ' options.unicode.unicode ' , ' options.unicode.unicode1 ' , ' options.unicode.unicode2 ' , ' options.unicode.unicode3 ' , ' options.unicode.unicode4 ' , ' options.unicode.unicode6 ' , ' options.unicode.unicode7 ' , ' options.unicodetoto ' ] )
assert dico [ ' options.unicode.unicode ' ] == [ ' test ' , ' trah ' ]
assert dico [ ' options.unicode.unicode1 ' ] == [ None , None ]
assert dico [ ' options.unicode.unicode2 ' ] == [ ' test ' , None ]
assert dico [ ' options.unicode.unicode3 ' ] [ 0 ] is None
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] , PropertiesOptionError )
assert dico [ ' options.unicode.unicode4 ' ] [ 0 ] is None
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode6 ' ] [ 0 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode6 ' ] [ 1 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode7 ' ] [ 0 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode7 ' ] [ 1 ] , PropertiesOptionError )
assert dico [ ' options.unicodetoto ' ] is None
del dico [ ' options.unicode.unicode3 ' ] [ 1 ]
del dico [ ' options.unicode.unicode3 ' ]
del dico [ ' options.unicode.unicode4 ' ] [ 1 ]
del dico [ ' options.unicode.unicode4 ' ]
del dico [ ' options.unicode.unicode6 ' ] [ 1 ]
del dico [ ' options.unicode.unicode6 ' ] [ 0 ]
del dico [ ' options.unicode.unicode7 ' ] [ 1 ]
del dico [ ' options.unicode.unicode7 ' ] [ 0 ]
#
await cfg . option ( ' options.unicodetoto ' ) . value . set ( ' test ' )
dico = await cfg . value . dict ( )
assert dico . keys ( ) == set ( [ ' options.unicode.unicode ' , ' options.unicode.unicode1 ' , ' options.unicode.unicode2 ' , ' options.unicode.unicode3 ' , ' options.unicode.unicode4 ' , ' options.unicode.unicode5 ' , ' options.unicode.unicode6 ' , ' options.unicode.unicode7 ' , ' options.unicodetoto ' ] )
assert dico [ ' options.unicode.unicode ' ] == [ ' test ' , ' trah ' ]
assert dico [ ' options.unicode.unicode1 ' ] == [ None , None ]
assert dico [ ' options.unicode.unicode2 ' ] == [ ' test ' , None ]
assert dico [ ' options.unicode.unicode3 ' ] [ 0 ] is None
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] , PropertiesOptionError )
assert dico [ ' options.unicode.unicode4 ' ] [ 0 ] is None
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] , PropertiesOptionError )
assert dico [ ' options.unicode.unicode5 ' ] == [ None , None ]
assert dico [ ' options.unicode.unicode6 ' ] [ 0 ] is None
assert isinstance ( dico [ ' options.unicode.unicode6 ' ] [ 1 ] , PropertiesOptionError )
assert dico [ ' options.unicode.unicode7 ' ] [ 0 ] is None
assert isinstance ( dico [ ' options.unicode.unicode7 ' ] [ 1 ] , PropertiesOptionError )
assert dico [ ' options.unicodetoto ' ] == ' test '
del dico [ ' options.unicode.unicode3 ' ] [ 1 ]
del dico [ ' options.unicode.unicode3 ' ]
del dico [ ' options.unicode.unicode4 ' ] [ 1 ]
del dico [ ' options.unicode.unicode4 ' ]
del dico [ ' options.unicode.unicode6 ' ] [ 1 ]
del dico [ ' options.unicode.unicode6 ' ] [ 0 ]
del dico [ ' options.unicode.unicode7 ' ] [ 1 ]
del dico [ ' options.unicode.unicode7 ' ] [ 0 ]
assert not await list_sessions ( )
@pytest.mark.asyncio
async def test_leadership_requires_transitive1 ( config_type ) :
optiontoto = StrOption ( ' unicodetoto ' , " Simple unicode " )
option = StrOption ( ' unicode ' , " Unicode leader " , multi = True )
option1 = StrOption ( ' unicode1 ' , " Unicode follower 1 " , multi = True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( optiontoto , raisepropertyerror = True ) ,
' expected ' : ParamValue ( ' test ' ) ,
' reverse_condition ' : ParamValue ( True ) } ) )
option2 = StrOption ( ' unicode2 ' , " Unicode follower 2 " , properties = ( disabled_property , ) , multi = True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( option2 , raisepropertyerror = True ) ,
' expected ' : ParamValue ( ' test ' ) ,
' index ' : ParamIndex ( ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' reverse_condition ' : ParamValue ( True ) } ) )
option3 = StrOption ( ' unicode3 ' , " Unicode follower 3 " , properties = ( disabled_property , ) , multi = True )
disabled_property = Calculation ( calc_value ,
Params ( ParamValue ( ' disabled ' ) ,
kwargs = { ' condition ' : ParamOption ( option3 , raisepropertyerror = True ) ,
' expected ' : ParamValue ( ' test ' ) ,
' index ' : ParamIndex ( ) ,
' no_condition_is_invalid ' : ParamValue ( True ) ,
' reverse_condition ' : ParamValue ( True ) } ) )
option4 = StrOption ( ' unicode4 ' , " Unicode follower 4 " , properties = ( disabled_property , ) , multi = True )
descr1 = Leadership ( " unicode " , " Common configuration 1 " ,
[ option , option1 , option2 , option3 , option4 ] )
descr = OptionDescription ( " options " , " Common configuration 2 " , [ descr1 , optiontoto ] )
descr = OptionDescription ( " unicode1 " , " " , [ descr ] )
async with await Config ( descr ) as cfg :
await cfg . property . read_write ( )
cfg = await get_config ( cfg , config_type )
assert await cfg . value . dict ( ) == { ' options.unicode.unicode ' : [ ] , ' options.unicode.unicode1 ' : [ ] , ' options.unicode.unicode3 ' : [ ] , ' options.unicode.unicode4 ' : [ ] , ' options.unicodetoto ' : None }
#
await cfg . option ( ' options.unicodetoto ' ) . value . set ( ' test ' )
assert await cfg . value . dict ( ) == { ' options.unicode.unicode ' : [ ] , ' options.unicode.unicode1 ' : [ ] , ' options.unicode.unicode2 ' : [ ] , ' options.unicode.unicode3 ' : [ ] , ' options.unicode.unicode4 ' : [ ] , ' options.unicodetoto ' : ' test ' }
#
await cfg . option ( ' options.unicode.unicode ' ) . value . set ( [ ' a ' , ' b ' , ' c ' ] )
dico = await cfg . value . dict ( )
assert list ( dico . keys ( ) ) == [ ' options.unicode.unicode ' , ' options.unicode.unicode1 ' , ' options.unicode.unicode2 ' , ' options.unicode.unicode3 ' , ' options.unicode.unicode4 ' , ' options.unicodetoto ' ]
assert dico [ ' options.unicodetoto ' ] == ' test '
assert dico [ ' options.unicode.unicode ' ] == [ ' a ' , ' b ' , ' c ' ]
assert dico [ ' options.unicode.unicode1 ' ] == [ None , None , None ]
assert dico [ ' options.unicode.unicode2 ' ] == [ None , None , None ]
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 0 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 2 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 0 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 2 ] , PropertiesOptionError )
del ( dico [ ' options.unicode.unicode3 ' ] [ 2 ] )
del ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] )
del ( dico [ ' options.unicode.unicode3 ' ] [ 0 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 2 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 0 ] )
await cfg . option ( ' options.unicode.unicode2 ' , 1 ) . value . set ( ' test ' )
await cfg . option ( ' options.unicode.unicode3 ' , 1 ) . value . set ( ' test ' )
dico = await cfg . value . dict ( )
assert list ( dico . keys ( ) ) == [ ' options.unicode.unicode ' , ' options.unicode.unicode1 ' , ' options.unicode.unicode2 ' , ' options.unicode.unicode3 ' , ' options.unicode.unicode4 ' , ' options.unicodetoto ' ]
assert dico [ ' options.unicodetoto ' ] == ' test '
assert dico [ ' options.unicode.unicode ' ] == [ ' a ' , ' b ' , ' c ' ]
assert dico [ ' options.unicode.unicode1 ' ] == [ None , None , None ]
assert dico [ ' options.unicode.unicode2 ' ] == [ None , ' test ' , None ]
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 0 ] , PropertiesOptionError )
assert dico [ ' options.unicode.unicode3 ' ] [ 1 ] == ' test '
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 2 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 0 ] , PropertiesOptionError )
assert dico [ ' options.unicode.unicode4 ' ] [ 1 ] == None
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 2 ] , PropertiesOptionError )
del ( dico [ ' options.unicode.unicode3 ' ] [ 2 ] )
del ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] )
del ( dico [ ' options.unicode.unicode3 ' ] [ 0 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 2 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 0 ] )
#
await cfg . option ( ' options.unicode.unicode2 ' , 1 ) . value . set ( ' rah ' )
dico = await cfg . value . dict ( )
assert list ( dico . keys ( ) ) == [ ' options.unicode.unicode ' , ' options.unicode.unicode1 ' , ' options.unicode.unicode2 ' , ' options.unicode.unicode3 ' , ' options.unicode.unicode4 ' , ' options.unicodetoto ' ]
assert dico [ ' options.unicodetoto ' ] == ' test '
assert dico [ ' options.unicode.unicode ' ] == [ ' a ' , ' b ' , ' c ' ]
assert dico [ ' options.unicode.unicode1 ' ] == [ None , None , None ]
assert dico [ ' options.unicode.unicode2 ' ] == [ None , ' rah ' , None ]
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 0 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 2 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 0 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 2 ] , PropertiesOptionError )
del ( dico [ ' options.unicode.unicode3 ' ] [ 2 ] )
del ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] )
del ( dico [ ' options.unicode.unicode3 ' ] [ 0 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 2 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 0 ] )
#
await cfg . option ( ' options.unicode.unicode2 ' , 1 ) . value . set ( ' test ' )
await cfg . option ( ' options.unicodetoto ' ) . value . set ( ' rah ' )
dico = await cfg . value . dict ( )
assert list ( dico . keys ( ) ) == [ ' options.unicode.unicode ' , ' options.unicode.unicode1 ' , ' options.unicode.unicode3 ' , ' options.unicode.unicode4 ' , ' options.unicodetoto ' ]
assert dico [ ' options.unicodetoto ' ] == ' rah '
assert dico [ ' options.unicode.unicode ' ] == [ ' a ' , ' b ' , ' c ' ]
assert dico [ ' options.unicode.unicode1 ' ] == [ None , None , None ]
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 0 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode3 ' ] [ 2 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 0 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] , PropertiesOptionError )
assert isinstance ( dico [ ' options.unicode.unicode4 ' ] [ 2 ] , PropertiesOptionError )
del ( dico [ ' options.unicode.unicode3 ' ] [ 2 ] )
del ( dico [ ' options.unicode.unicode3 ' ] [ 1 ] )
del ( dico [ ' options.unicode.unicode3 ' ] [ 0 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 2 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 1 ] )
del ( dico [ ' options.unicode.unicode4 ' ] [ 0 ] )
assert not await list_sessions ( )
# FIXME tester l'ajout d'un Calculation
# FIXME permissive peut etre in calcul !
# FIXME Calculation sur des multis ...