stove/tiramisu
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

add allow_protocol to PortOption

Browse source
This commit is contained in:
Emmanuel Garette 2022-10-01 19:43:49 +02:00
parent 89c095e0d3
commit 22ef5c7c3b
3 changed files with 26 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
tests/test_config_api.py
View file

@ -371,6 +371,8 @@ async def test_invalid_option():
PortOption('a', '', allow_zero=True, allow_wellknown=False, allow_registred=True, allow_private=True) PortOption('a', '', allow_zero=True, allow_wellknown=False, allow_registred=True, allow_private=True)
with pytest.raises(ValueError): with pytest.raises(ValueError):
PortOption('a', '', allow_zero=False, allow_wellknown=False, allow_registred=False, allow_private=False) PortOption('a', '', allow_zero=False, allow_wellknown=False, allow_registred=False, allow_private=False)
with pytest.raises(ValueError):
PortOption('a', '', 'tcp:80')
NetworkOption('a', '') NetworkOption('a', '')
with pytest.raises(ValueError): with pytest.raises(ValueError):
NetworkOption('a', '', 'string') NetworkOption('a', '', 'string')

10
tests/test_config_ip.py
View file

@ -295,6 +295,16 @@ async def test_port(config_type):
assert not await list_sessions() assert not await list_sessions()
@pytest.mark.asyncio
async def test_port_protocol(config_type):
a = PortOption('a', '', allow_protocol=True)
od = OptionDescription('od', '', [a])
async with await Config(od) as cfg:
await cfg.option('a').value.set('80')
await cfg.option('a').value.set('tcp:80')
assert not await list_sessions()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_port_range(config_type): async def test_port_range(config_type):
a = PortOption('a', '', allow_range=True) a = PortOption('a', '', allow_range=True)

17
tiramisu/option/portoption.py
View file

@ -48,12 +48,15 @@ class PortOption(StrOption):
allow_zero: bool=False, allow_zero: bool=False,
allow_wellknown: bool=True, allow_wellknown: bool=True,
allow_registred: bool=True, allow_registred: bool=True,
allow_protocol: bool=False,
allow_private: bool=False, allow_private: bool=False,
**kwargs) -> None: **kwargs) -> None:
extra = {'_allow_range': allow_range, extra = {'_allow_range': allow_range,
'_allow_protocol': allow_protocol,
'_min_value': None, '_min_value': None,
'_max_value': None} '_max_value': None,
}
ports_min = [0, 1, 1024, 49152] ports_min = [0, 1, 1024, 49152]
ports_max = [0, 1023, 49151, 65535] ports_max = [0, 1023, 49151, 65535]
is_finally = False is_finally = False
@ -81,7 +84,9 @@ class PortOption(StrOption):
def validate(self, def validate(self,
value: str) -> None: value: str) -> None:
super().validate(value) super().validate(value)
if self.impl_get_extra('_allow_range') and ":" in str(value): if self.impl_get_extra('_allow_protocol') and (value.startswith('tcp:') or value.startswith('udp:')):
value = [value[4:]]
elif self.impl_get_extra('_allow_range') and ":" in str(value):
value = value.split(':') value = value.split(':')
if len(value) != 2: if len(value) != 2:
raise ValueError(_('range must have two values only')) raise ValueError(_('range must have two values only'))
@ -98,7 +103,13 @@ class PortOption(StrOption):
def second_level_validation(self, def second_level_validation(self,
value: str, value: str,
warnings_only: bool) -> None: warnings_only: bool) -> None:
for val in value.split(':'): if self.impl_get_extra('_allow_protocol') and (value.startswith('tcp:') or value.startswith('udp:')):
value = [value[4:]]
elif ':' in value:
value = value.split(':')
else:
value = [value]
for val in value:
val = int(val) val = int(val)
if not self.impl_get_extra('_min_value') <= val <= self.impl_get_extra('_max_value'): if not self.impl_get_extra('_min_value') <= val <= self.impl_get_extra('_max_value'):
if warnings_only: if warnings_only: