tiramisu/tiramisu/function.py
2023-05-11 15:44:48 +02:00

597 lines
26 KiB
Python

# Copyright (C) 2018-2023 Team tiramisu (see AUTHORS for all contributors)
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""some functions to validates or calculates value
"""
from typing import Any, List, Optional
from operator import add, mul, sub, truediv
from ipaddress import ip_address, ip_interface, ip_network
from .i18n import _
from .setting import undefined
from .error import display_list
FUNCTION_WAITING_FOR_DICT = []
def function_waiting_for_dict(function):
"""functions (calculation or validation) receive by default only the value of other options
all functions declared with this function recieve a dict with option informations
(value, name, ...)
"""
name = function.__name__
if name not in FUNCTION_WAITING_FOR_DICT:
FUNCTION_WAITING_FOR_DICT.append(name)
return function
@function_waiting_for_dict
def valid_network_netmask(network: dict,
netmask: dict,
):
"""
validates if network and netmask are coherent
this validator must be set to netmask option
"""
if None in [network['value'], netmask['value']]:
return
try:
ip_network(f'{network["value"]}/{netmask["value"]}')
except ValueError as err:
raise ValueError(_(f'network "{network["value"]}" ({network["name"]}) does not match '
'with this netmask')) from err
@function_waiting_for_dict
def valid_ip_netmask(ip: dict, # pylint: disable=invalid-name
netmask: dict,
):
"""validates if ip and netmask are coherent
this validator must be set to netmask option
"""
if None in [ip['value'], netmask['value']]:
return
ip_netmask = ip_interface(f'{ip["value"]}/{netmask["value"]}')
if ip_netmask.ip == ip_netmask.network.network_address:
msg = _(f'IP "{ip["value"]}" ({ip["name"]}) with this netmask is '
'in fact a network address')
raise ValueError(msg)
if ip_netmask.ip == ip_netmask.network.broadcast_address:
msg = _(f'IP "{ip["value"]}" ({ip["name"]}) with this netmask is '
'in fact a broacast address')
raise ValueError(msg)
@function_waiting_for_dict
def valid_broadcast(network: dict,
netmask: dict,
broadcast: dict,
):
"""validates if the broadcast is coherent with network and netmask
"""
if None in [network['value'], netmask['value'], broadcast['value']]:
return
if ip_network(f'{network["value"]}/{netmask["value"]}').broadcast_address != \
ip_address(broadcast['value']):
msg = _(f'broadcast invalid with network {network["value"]} ({network["name"]}) '
f'and netmask {netmask["value"]} ({netmask["name"]})')
raise ValueError(msg)
@function_waiting_for_dict
def valid_in_network(ip: dict, # pylint: disable=invalid-name
network: dict,
netmask=Optional[dict],
):
"""validates if an IP is in a network
this validator must be set to ip option
"""
if None in [ip['value'], network['value']]:
return
if '/' in network['value']:
# it's a CIDR network
network_value = network['value']
else:
if netmask is None or netmask['value'] is None:
return
network_value = f'{network["value"]}/{netmask["value"]}'
network_obj = ip_network(network_value)
ip_netmask = ip_interface(f'{ip["value"]}/{network_obj.netmask}')
if ip_netmask not in network_obj:
if netmask is None:
msg = _('this IP is not in network {network["value"]} ({network["name"]})')
else:
msg = _('this IP is not in network {network["value"]} ({network["name"]}) '
'with netmask {netmask["value"]} ({netmask["name"]})')
raise ValueError(msg)
# test if ip is not network/broadcast IP
if ip_netmask.ip == ip_netmask.network.network_address:
msg = _(f'this IP with the network {network["value"]} ({network["value"]} '
'is in fact a network address')
raise ValueError(msg)
if ip_netmask.ip == ip_netmask.network.broadcast_address:
msg = _(f'this IP with the network {network["value"]} ({network["value"]} '
'is in fact a broadcast address')
raise ValueError(msg)
@function_waiting_for_dict
def valid_not_equal(*values):
"""valid that two options have not same value
"""
equal = set()
for val in values[1:]:
if 'propertyerror' in val:
continue
if values[0]['value'] == val['value'] is not None:
equal.add(val['name'])
if not equal:
return
msg = _(f'value is identical to {display_list(list(equal), add_quote=True)}')
raise ValueError(msg)
class CalcValue:
"""class to calc_value with different functions
"""
# pylint: disable=too-many-instance-attributes
def __call__(self,
*args: List[Any],
multi: bool=False,
default: Any=undefined,
condition: Any=undefined,
no_condition_is_invalid: bool=False,
expected: Any=undefined,
condition_operator: str='AND',
reverse_condition: bool=False,
allow_none: bool=False,
remove_duplicate_value: bool=False,
join: Optional[str]=None,
min_args_len: Optional[int]=None,
operator: Optional[str]=None,
index: Optional[int]=None,
**kwargs) -> Any:
# pylint: disable=too-many-statements,too-many-branches,too-many-nested-blocks,too-many-locals
"""calculate value
:param args: list of value
:param multi: value returns must be a list of value
:param default: default value if condition is not matched or if args is empty
if there is more than one default value, set default_0, default_1, ...
:param condition: test if condition is equal to expected value
if there is more than one condition, set condition_0, condition_1, ...
:param expected: value expected for all conditions
if expected value is different between condition, set expected_0,
expected_1, ...
:param no_condition_is_invalid: if no condition and not condition_0, condition_1, ... (for
example if option is disabled) consider that condition not
matching
:param condition_operator: OR or AND operator for condition
:param allow_none: if False, do not return list in None is present in list
:param remove_duplicate_value: if True, remote duplicated value
:param join: join all args with specified characters
:param min_args_len: if number of arguments is smaller than this value, return default value
:param operator: 'add', 'mul', 'div' or 'sub' all args (args must be integer value)
:param index: index for follower
examples:
* you want to copy value from an option to an other option:
>>> from tiramisu import calc_value, StrOption, OptionDescription, Config, \
... Params, ParamOption
>>> val1 = StrOption('val1', '', 'val1')
>>> val2 = StrOption('val2', '', callback=calc_value,
... callback_params=Params(ParamOption(val1)))
>>> od = OptionDescription('root', '', [val1, val2])
>>> cfg = Config(od)
>>> cfg.value.dict()
{'val1': 'val1', 'val2': 'val1'}
* you want to copy values from two options in one multi option
>>> from tiramisu import calc_value, StrOption, OptionDescription, Config, Params, \
... ParamOption, ParamValue
>>> val1 = StrOption('val1', "", 'val1')
>>> val2 = StrOption('val2', "", 'val2')
>>> val3 = StrOption('val3', "", multi=True, callback=calc_value,
... callback_params=Params((ParamOption(val1), ParamOption(val2)),
... multi=ParamValue(True)))
>>> od = OptionDescription('root', '', [val1, val2, val3])
>>> cfg = Config(od)
>>> cfg.value.dict()
{'val1': 'val1', 'val2': 'val2', 'val3': ['val1', 'val2']}
* you want to copy a value from an option if it not disabled, otherwise set 'default_value'
>>> from tiramisu import calc_value, StrOption, OptionDescription, Config, Params, \
... ParamOption, ParamValue
>>> val1 = StrOption('val1', '', 'val1')
>>> val2 = StrOption('val2', '', callback=calc_value,
... callback_params=Params(ParamOption(val1, True),
... default=ParamValue('default_value')))
>>> od = OptionDescription('root', '', [val1, val2])
>>> cfg = Config(od)
>>> cfg.property.read_write()
>>> cfg.value.dict()
{'val1': 'val1', 'val2': 'val1'}
>>> cfg.option('val1').property.add('disabled')
>>> cfg.value.dict()
{'val2': 'default_value'}
* you want to copy value from an option if an other is True, otherwise set 'default_value'
>>> from tiramisu import calc_value, BoolOption, StrOption, OptionDescription, Config, \
... Params, ParamOption, ParamValue
>>> boolean = BoolOption('boolean', '', True)
>>> val1 = StrOption('val1', '', 'val1')
>>> val2 = StrOption('val2', '', callback=calc_value,
... callback_params=Params(ParamOption(val1, True),
... default=ParamValue('default_value'),
... condition=ParamOption(boolean),
... expected=ParamValue(True)))
>>> od = OptionDescription('root', '', [boolean, val1, val2])
>>> cfg = Config(od)
>>> cfg.property.read_write()
>>> cfg.value.dict()
{'boolean': True, 'val1': 'val1', 'val2': 'val1'}
>>> cfg.option('boolean').value.set(False)
>>> cfg.value.dict()
{'boolean': False, 'val1': 'val1', 'val2': 'default_value'}
* you want to copy option even if None is present
>>> from tiramisu import calc_value, StrOption, OptionDescription, Config, Params, \
... ParamOption, ParamValue
>>> val1 = StrOption('val1', "", 'val1')
>>> val2 = StrOption('val2', "")
>>> val3 = StrOption('val3', "", multi=True, callback=calc_value,
... callback_params=Params((ParamOption(val1), ParamOption(val2)),
... multi=ParamValue(True), allow_none=ParamValue(True)))
>>> od = OptionDescription('root', '', [val1, val2, val3])
>>> cfg = Config(od)
>>> cfg.value.dict()
{'val1': 'val1', 'val2': None, 'val3': ['val1', None]}
* you want uniq value
>>> from tiramisu import calc_value, StrOption, OptionDescription, Config, Params, \
... ParamOption, ParamValue
>>> val1 = StrOption('val1', "", 'val1')
>>> val2 = StrOption('val2', "", 'val1')
>>> val3 = StrOption('val3', "", multi=True, callback=calc_value,
... callback_params=Params((ParamOption(val1), ParamOption(val2)),
... multi=ParamValue(True), remove_duplicate_value=ParamValue(True)))
>>> od = OptionDescription('root', '', [val1, val2, val3])
>>> cfg = Config(od)
>>> cfg.value.dict()
{'val1': 'val1', 'val2': 'val1', 'val3': ['val1']}
* you want to join two values with '.'
>>> from tiramisu import calc_value, StrOption, OptionDescription, Config, Params, \
... ParamOption, ParamValue
>>> val1 = StrOption('val1', "", 'val1')
>>> val2 = StrOption('val2', "", 'val2')
>>> val3 = StrOption('val3', "", callback=calc_value,
... callback_params=Params((ParamOption(val1),
... ParamOption(val2)), join=ParamValue('.')))
>>> od = OptionDescription('root', '', [val1, val2, val3])
>>> cfg = Config(od)
>>> cfg.value.dict()
{'val1': 'val1', 'val2': 'val2', 'val3': 'val1.val2'}
* you want join three values, only if almost three values are set
>>> from tiramisu import calc_value, StrOption, OptionDescription, Config, Params, \
... ParamOption, ParamValue
>>> val1 = StrOption('val1', "", 'val1')
>>> val2 = StrOption('val2', "", 'val2')
>>> val3 = StrOption('val3', "", 'val3')
>>> val4 = StrOption('val4', "", callback=calc_value,
... callback_params=Params((ParamOption(val1),
... ParamOption(val2),
... ParamOption(val3, True)),
... join=ParamValue('.'), min_args_len=ParamValue(3)))
>>> od = OptionDescription('root', '', [val1, val2, val3, val4])
>>> cfg = Config(od)
>>> cfg.property.read_write()
>>> cfg.value.dict()
{'val1': 'val1', 'val2': 'val2', 'val3': 'val3', 'val4': 'val1.val2.val3'}
>>> cfg.option('val3').property.add('disabled')
>>> cfg.value.dict()
{'val1': 'val1', 'val2': 'val2', 'val4': ''}
* you want to add all values
>>> from tiramisu import calc_value, IntOption, OptionDescription, Config, Params, \
... ParamOption, ParamValue
>>> val1 = IntOption('val1', "", 1)
>>> val2 = IntOption('val2', "", 2)
>>> val3 = IntOption('val3', "", callback=calc_value,
... callback_params=Params((ParamOption(val1),
ParamOption(val2)),
... operator=ParamValue('add')))
>>> od = OptionDescription('root', '', [val1, val2, val3])
>>> cfg = Config(od)
>>> cfg.value.dict()
{'val1': 1, 'val2': 2, 'val3': 3}
"""
# pylint: disable=attribute-defined-outside-init
self.args = args
self.condition = condition
self.expected = expected
self.condition_operator = condition_operator
self.reverse_condition = reverse_condition
self.kwargs = kwargs
self.no_condition_is_invalid = no_condition_is_invalid # pylint: disable=attribute-defined-outside-init
value = self.get_value(default,
min_args_len,
)
if not multi:
if join is not None:
if None not in value:
value = join.join(value)
else:
value = None
elif value and operator:
new_value = value[0]
oper = {'mul': mul,
'add': add,
'div': truediv,
'sub': sub,
}[operator]
for val in value[1:]:
new_value = oper(new_value, val)
value = new_value
elif value == []:
value = None
else:
value = value[0]
if isinstance(value, list) and index is not None:
if len(value) > index:
value = value[index]
else:
value = None
else:
if join is not None:
if None not in value:
length_val = None
for val in value:
if isinstance(val, list):
if None in val:
length_val = None
break
lval = len(val)
if length_val is not None and length_val != lval:
msg = _('unexpected value in calc_value with join attribute '
f'"{val}" with invalid length "{length_val}"')
raise ValueError(msg)
length_val = lval
new_value = []
if length_val is not None:
for idx in range(length_val):
idx_val = []
for val in value:
if isinstance(val, list):
idx_val.append(val[idx])
else:
idx_val.append(val)
new_value.append(join.join(idx_val))
value = new_value
else:
value = []
elif None in value and not allow_none:
value = []
if remove_duplicate_value:
new_value = []
for val in value:
if val not in new_value:
new_value.append(val)
value = new_value
return value
def value_from_kwargs(self,
value: Any,
pattern: str,
to_dict: bool=False,
empty_test=undefined) -> Any:
"""get value from kwargs
"""
# pylint: disable=too-many-branches
# if value attribute exist return it's value
# otherwise pattern_0, pattern_1, ...
# otherwise undefined
if value is not empty_test:
if to_dict == 'all':
returns = {None: value}
else:
returns = value
else:
kwargs_matches = {}
len_pattern = len(pattern)
for key, pattern_value in self.kwargs.items():
if key.startswith(pattern):
index = int(key[len_pattern:])
if isinstance(pattern_value, dict):
pattern_value = pattern_value['value']
kwargs_matches[index] = pattern_value
if not kwargs_matches:
returns = undefined
else:
keys = sorted(kwargs_matches)
if to_dict:
returns = {}
else:
returns = []
for key in keys:
if to_dict:
returns[key] = kwargs_matches[key]
else:
returns.append(kwargs_matches[key])
return returns
def is_condition_matches(self,
condition_value,
):
"""verify the condition
"""
# pylint: disable=too-many-branches
calculated_conditions = self.value_from_kwargs(condition_value,
'condition_',
to_dict='all',
)
if calculated_conditions is undefined:
is_matches = not self.no_condition_is_invalid
else:
is_matches = None
calculated_expected = self.value_from_kwargs(self.expected,
'expected_',
to_dict=True,
)
calculated_reverse = self.value_from_kwargs(self.reverse_condition,
'reverse_condition_',
to_dict=True,
empty_test=False,
)
for idx, calculated_condition in calculated_conditions.items():
if isinstance(calculated_expected, dict):
if idx is not None:
if isinstance(calculated_expected[idx], list):
current_matches = calculated_condition in calculated_expected[idx]
else:
current_matches = calculated_condition == calculated_expected[idx]
else:
current_matches = calculated_condition in calculated_expected.values()
else:
current_matches = calculated_condition == calculated_expected
if isinstance(calculated_reverse, dict) and idx in calculated_reverse:
reverse_condition = calculated_reverse[idx]
else:
reverse_condition = False
if is_matches is None:
is_matches = current_matches
if self.condition_operator == 'AND':
is_matches = is_matches and current_matches
if reverse_condition:
is_matches = not is_matches
if not is_matches:
break
elif self.condition_operator == 'OR':
is_matches = is_matches or current_matches
if reverse_condition:
is_matches = not is_matches
if is_matches:
break
else:
msg = _(f'unexpected {self.condition_operator} condition_operator '
'in calc_value')
raise ValueError(msg)
is_matches = is_matches and not self.reverse_condition \
or not is_matches and self.reverse_condition
return is_matches
def get_value(self,
default,
min_args_len,
):
"""get the value from arguments
"""
# retrieve the condition
if isinstance(self.condition, dict):
if 'value' in self.condition:
condition_value = self.condition['value']
else:
condition_value = undefined
else:
condition_value = self.condition
# value is empty if condition doesn't match
# otherwise value is arg
if not self.is_condition_matches(condition_value):
value = []
else:
value = self.get_args()
if min_args_len and not len(value) >= min_args_len:
value = []
if not value:
# default value
new_default = self.value_from_kwargs(default,
'default_',
)
if new_default is not undefined:
if not isinstance(new_default, list):
value = [new_default]
else:
value = new_default
return value
def get_args(self):
"""get all arguments
"""
return list(self.args)
class CalcValuePropertyHelp(CalcValue):
"""special class to display property error
"""
def get_name(self):
"""get the condition name
"""
return self.condition['name']
def get_indexed_name(self, index: int) -> str:
"""get name for a specified index
"""
condition_index = self.kwargs.get(f'condition_{index}')
if condition_index is not None and not isinstance(condition_index, dict):
raise ValueError(_(f'unexpected condition_{index} must have "todict" argument'))
return condition_index['name']
def build_property_message(self,
name: str,
value: Any,
) -> str:
"""prepare message to display error message if needed
"""
if not self.reverse_condition:
msg = _('the value of "{0}" is {1}').format(name, value)
else:
msg = _('the value of "{0}" is not {1}').format(name, value)
return msg
def get_args(self):
args = super().get_args()
action = args[0]
calculated_expected = self.value_from_kwargs(self.expected,
'expected_',
to_dict=True)
if self.condition is not undefined:
if 'propertyerror' in self.condition:
msg = self.condition['propertyerror']
else:
name = self.get_name()
if isinstance(calculated_expected, dict):
calc_values = calculated_expected.values()
else:
calc_values = [calculated_expected]
display_value = display_list([str(val) for val in calc_values],
'or',
add_quote=True)
msg = self.build_property_message(name, display_value)
else:
msgs = []
for key, value in calculated_expected.items():
name = self.get_indexed_name(key)
msgs.append(self.build_property_message(name, f'"{value}"'))
msg = display_list(msgs, self.condition_operator.lower())
return [(action, f'"{action}" ({msg})')]
calc_value = CalcValue()
calc_value.__name__ = 'calc_value' # pylint: disable=attribute-defined-outside-init
# function_waiting_for_dict(calc_value)
calc_value_property_help = CalcValuePropertyHelp()
calc_value_property_help.__name__ = 'calc_value_property_help' # pylint: disable=attribute-defined-outside-init
function_waiting_for_dict(calc_value_property_help)