in source/code/configuration/task_configuration.py [0:0]
def verify_task_parameters(self, task_parameters, task_settings, action_name):
"""
Validates parameter set and values for the specified action.
A ValueException is raised in the following cases:
-Required parameter is not available and there is no default specified for the action
-Unknown parameter found
-Type of parameter is wrong and value can task_parameters not be converted to the required type
-Value is out of range specified by min and max value for numeric parameters
-Value is too long, too short or does not mats the allowed pattern for string parameters
:param task_parameters: Dictionary of parameters, keys are name of the parameters, value is parameter value
:param task_settings: Task settings without parameters included yet
:param action_name: Name of the action
:return: Dictionary of validated parameters, missing non required parameters are set to default if specified in action
implementation
"""
validated_parameters = {}
def verify_numeric_parameter(value, action_param):
if type(value) in [int, float, int, complex, decimal]:
if actions.PARAM_MIN_VALUE in action_param and value < action_param[actions.PARAM_MIN_VALUE]:
raise_value_error(ERR_MIN_VALUE, value, action_param[actions.PARAM_MIN_VALUE], param_name)
if actions.PARAM_MAX_VALUE in action_param and value > action_param[actions.PARAM_MAX_VALUE]:
raise_value_error(ERR_MAX_VALUE, value, action_param[actions.PARAM_MAX_VALUE], param_name)
def verify_string_parameter(value, action_param):
if type(value) in [str, str]:
if actions.PARAM_MIN_LEN in action_param and len(value) < action_param[actions.PARAM_MIN_LEN]:
raise_value_error(ERR_MIN_LEN, value, action_param[actions.PARAM_MIN_LEN], param_name)
if actions.PARAM_MAX_LEN in action_param and len(value) > action_param[actions.PARAM_MAX_LEN]:
raise_value_error(ERR_MAX_LEN, value, action_param[actions.PARAM_MAX_LEN], param_name)
if actions.PARAM_PATTERN in action_param and not re.match(action_param[actions.PARAM_PATTERN], value):
raise_value_error(ERR_PATTERN_VALUE, value, action_param[actions.PARAM_PATTERN], param_name)
def verify_known_parameter(parameters, action_params):
# test for unknown parameters in the task definition
for tp in parameters:
if tp not in action_params:
if len(action_params) > 0:
self._logger.warning(WARN_INVALID_PARAMETER, tp, action_name, ", ".join(action_params))
else:
self._logger.warning(WARN_NO_PARAMETERS, tp, action_name)
def verify_parameter_type(value, action_param):
parameter_type = action_param.get(actions.PARAM_TYPE)
if parameter_type is not None:
if type(value) != parameter_type:
try:
# value does not match type, try to convert
if parameter_type == bool:
return TaskConfiguration.as_boolean(str(value))
return parameter_type(value)
except Exception:
# not possible to convert to required type
raise ValueError(
ERR_WRONG_PARAM_TYPE.format(param_name, str(parameter_type), parameter_value,
type(parameter_value)))
return value
def verify_allowed_values(value, action_param):
if actions.PARAM_ALLOWED_VALUES in action_param and value not in action_param[actions.PARAM_ALLOWED_VALUES]:
raise ValueError(
ERR_NOT_ALLOWED_VALUE.format(str(parameter_value), ",".join(action_param[actions.PARAM_ALLOWED_VALUES]),
param_name))
def verify_required_parameter_available(parameter_name, action_params, parameters):
if action_params[parameter_name].get(actions.PARAM_REQUIRED, False) and parameter_name not in parameters:
raise_value_error(ERR_REQUIRED_PARAM_MISSING, parameter_name)
def get_param_value(name, action_param, parameters):
value = parameters.get(name)
if value is None:
value = action_param.get(actions.PARAM_DEFAULT)
return value
def action_class_parameter_check(parameters, tsk_settings, name):
# get the class that implements the action and test if there is a static method for additional checks of the parameters
action_class = actions.get_action_class(name)
validate_params_method = getattr(action_class, handlers.ACTION_VALIDATE_PARAMETERS_METHOD, None)
# if the method exists then validate the parameters using the business logic for that class
try:
if validate_params_method is not None:
return validate_params_method(parameters, tsk_settings, self._logger)
except Exception as ex:
self._logger.error(ERR_VALIDATING_TASK_PARAMETERS, name, ex)
raise_value_error(ERR_VALIDATING_TASK_PARAMETERS, name, ex)
return parameters
action_properties = actions.get_action_properties(action_name)
action_parameters = action_properties.get(actions.ACTION_PARAMETERS, {})
verify_known_parameter(task_parameters, action_parameters)
for param_name in action_parameters:
verify_required_parameter_available(param_name, action_parameters, task_parameters)
action_parameter = action_parameters[param_name]
parameter_value = get_param_value(param_name, action_parameter, task_parameters)
if parameter_value is not None:
parameter_value = verify_parameter_type(parameter_value, action_parameter)
verify_allowed_values(parameter_value, action_parameter)
verify_numeric_parameter(parameter_value, action_parameter)
verify_string_parameter(parameter_value, action_parameter)
validated_parameters[param_name] = parameter_value
validated_parameters = action_class_parameter_check(parameters=validated_parameters, tsk_settings=task_settings,
name=action_name)
return validated_parameters