diff --git a/lib/galaxy/tool_util/parameters/_types.py b/lib/galaxy/tool_util/parameters/_types.py index 2b97ee16c200..64fa9a3274fc 100644 --- a/lib/galaxy/tool_util/parameters/_types.py +++ b/lib/galaxy/tool_util/parameters/_types.py @@ -6,6 +6,7 @@ """ from typing import ( + Any, cast, List, Optional, @@ -15,6 +16,7 @@ # https://stackoverflow.com/questions/56832881/check-if-a-field-is-typing-optional from typing_extensions import ( + Annotated, get_args, get_origin, ) @@ -46,3 +48,12 @@ def cast_as_type(arg) -> Type: def is_optional(field) -> bool: return get_origin(field) is Union and type(None) in get_args(field) + + +def expand_annotation(field: Type, new_annotations: List[Any]) -> Type: + is_annotation = get_origin(field) is Annotated + if is_annotation: + args = get_args(field) # noqa: F841 + return Annotated[tuple([args[0], *args[1:], *new_annotations])] # type: ignore[return-value] + else: + return Annotated[tuple([field, *new_annotations])] # type: ignore[return-value] diff --git a/lib/galaxy/tool_util/parameters/factory.py b/lib/galaxy/tool_util/parameters/factory.py index e200b67aeb0e..ef731dfa8e60 100644 --- a/lib/galaxy/tool_util/parameters/factory.py +++ b/lib/galaxy/tool_util/parameters/factory.py @@ -14,6 +14,15 @@ PagesSource, ToolSource, ) +from galaxy.tool_util.parser.parameter_validators import ( + EmptyFieldParameterValidatorModel, + ExpressionParameterValidatorModel, + InRangeParameterValidatorModel, + LengthParameterValidatorModel, + NoOptionsParameterValidatorModel, + RegexParameterValidatorModel, + static_validators, +) from galaxy.tool_util.parser.util import parse_profile_version from galaxy.util import string_as_bool from .models import ( @@ -42,10 +51,13 @@ HiddenParameterModel, IntegerParameterModel, LabelValue, + NumberCompatiableValidators, RepeatParameterModel, RulesParameterModel, SectionParameterModel, + SelectCompatiableValidators, SelectParameterModel, + TextCompatiableValidators, TextParameterModel, ToolParameterBundle, ToolParameterBundleModel, @@ -82,7 +94,14 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool int_value = None else: raise ParameterDefinitionError() - return IntegerParameterModel(name=input_source.parse_name(), optional=optional, value=int_value) + static_validator_models = static_validators(input_source.parse_validators()) + validators: List[NumberCompatiableValidators] = [] + for static_validator in static_validator_models: + if static_validator.type == "in_range": + validators.append(cast(InRangeParameterValidatorModel, static_validator)) + return IntegerParameterModel( + name=input_source.parse_name(), optional=optional, value=int_value, validators=validators + ) elif param_type == "boolean": nullable = input_source.parse_optional() value = input_source.get_bool_or_none("checked", None if nullable else False) @@ -93,9 +112,21 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool ) elif param_type == "text": optional = input_source.parse_optional() + static_validator_models = static_validators(input_source.parse_validators()) + validators: List[TextCompatiableValidators] = [] + for static_validator in static_validator_models: + if static_validator.type == "length": + validators.append(cast(LengthParameterValidatorModel, static_validator)) + elif static_validator.type == "regex": + validators.append(cast(RegexParameterValidatorModel, static_validator)) + elif static_validator.type == "expression": + validators.append(cast(ExpressionParameterValidatorModel, static_validator)) + elif static_validator.type == "empty_field": + validators.append(cast(EmptyFieldParameterValidatorModel, static_validator)) return TextParameterModel( name=input_source.parse_name(), optional=optional, + validators=validators, ) elif param_type == "float": optional = input_source.parse_optional() @@ -107,10 +138,16 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool float_value = None else: raise ParameterDefinitionError() + static_validator_models = static_validators(input_source.parse_validators()) + validators: List[NumberCompatiableValidators] = [] + for static_validator in static_validator_models: + if static_validator.type == "in_range": + validators.append(cast(InRangeParameterValidatorModel, static_validator)) return FloatParameterModel( name=input_source.parse_name(), optional=optional, value=float_value, + validators=validators, ) elif param_type == "hidden": optional = input_source.parse_optional() @@ -158,11 +195,17 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool options = [] for option_label, option_value, selected in input_source.parse_static_options(): options.append(LabelValue(label=option_label, value=option_value, selected=selected)) + static_validator_models = static_validators(input_source.parse_validators()) + validators: List[SelectCompatiableValidators] = [] + for static_validator in static_validator_models: + if static_validator.type == "no_options": + validators.append(cast(NoOptionsParameterValidatorModel, static_validator)) return SelectParameterModel( name=input_source.parse_name(), optional=optional, options=options, multiple=multiple, + validators=validators, ) elif param_type == "drill_down": multiple = input_source.get_bool("multiple", False) diff --git a/lib/galaxy/tool_util/parameters/models.py b/lib/galaxy/tool_util/parameters/models.py index 7a70c4d5c245..9b17f3594ce5 100644 --- a/lib/galaxy/tool_util/parameters/models.py +++ b/lib/galaxy/tool_util/parameters/models.py @@ -24,6 +24,7 @@ Field, field_validator, HttpUrl, + PlainValidator, RootModel, StrictBool, StrictFloat, @@ -44,8 +45,18 @@ JsonTestCollectionDefDict, JsonTestDatasetDefDict, ) +from galaxy.tool_util.parser.parameter_validators import ( + EmptyFieldParameterValidatorModel, + ExpressionParameterValidatorModel, + InRangeParameterValidatorModel, + LengthParameterValidatorModel, + NoOptionsParameterValidatorModel, + RegexParameterValidatorModel, + StaticValidatorModel, +) from ._types import ( cast_as_type, + expand_annotation, is_optional as is_python_type_optional, list_type, optional, @@ -179,11 +190,37 @@ class LabelValue(BaseModel): selected: bool +TextCompatiableValidators = Union[ + LengthParameterValidatorModel, + RegexParameterValidatorModel, + ExpressionParameterValidatorModel, + EmptyFieldParameterValidatorModel, +] + + +def pydantic_validator_for(validator_model: StaticValidatorModel): + + def validator(v: Any) -> Any: + validator_model.statically_validate(v) + return v + + return PlainValidator(validator) + + +# actual typing will require generics and a typevar I think... +def static_tool_validators_to_pydantic(static_tool_param_validators: List[StaticValidatorModel]): + pydantic_validators = [] + for static_validator in static_tool_param_validators: + pydantic_validators.append(pydantic_validator_for(static_validator)) + return pydantic_validators + + class TextParameterModel(BaseGalaxyToolParameterModelDefinition): parameter_type: Literal["gx_text"] = "gx_text" area: bool = False default_value: Optional[str] = Field(default=None, alias="value") default_options: List[LabelValue] = [] + validators: List[TextCompatiableValidators] = [] @property def py_type(self) -> Type: @@ -196,6 +233,9 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam requires_value = self.request_requires_value if state_representation == "job_internal": requires_value = True + validators = static_tool_validators_to_pydantic(self.validators) + if validators: + py_type = expand_annotation(py_type, validators) return dynamic_model_information_from_py_type(self, py_type, requires_value=requires_value) @property @@ -203,12 +243,16 @@ def request_requires_value(self) -> bool: return False +NumberCompatiableValidators = Union[InRangeParameterValidatorModel,] + + class IntegerParameterModel(BaseGalaxyToolParameterModelDefinition): parameter_type: Literal["gx_integer"] = "gx_integer" optional: bool value: Optional[int] = None min: Optional[int] = None max: Optional[int] = None + validators: List[NumberCompatiableValidators] = None @property def py_type(self) -> Type: @@ -223,6 +267,9 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam requires_value = True elif _is_landing_request(state_representation): requires_value = False + validators = static_tool_validators_to_pydantic(self.validators) + if validators: + py_type = expand_annotation(py_type, validators) return dynamic_model_information_from_py_type(self, py_type, requires_value=requires_value) @property @@ -235,6 +282,7 @@ class FloatParameterModel(BaseGalaxyToolParameterModelDefinition): value: Optional[float] = None min: Optional[float] = None max: Optional[float] = None + validators: List[NumberCompatiableValidators] = None @property def py_type(self) -> Type: @@ -244,6 +292,9 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam py_type = self.py_type if state_representation == "workflow_step_linked": py_type = allow_connected_value(py_type) + validators = static_tool_validators_to_pydantic(self.validators) + if validators: + py_type = expand_annotation(py_type, validators) return dynamic_model_information_from_py_type(self, py_type) @property @@ -654,10 +705,14 @@ def request_requires_value(self) -> bool: return True +SelectCompatiableValidators = Union[NoOptionsParameterValidatorModel,] + + class SelectParameterModel(BaseGalaxyToolParameterModelDefinition): parameter_type: Literal["gx_select"] = "gx_select" options: Optional[List[LabelValue]] = None multiple: bool + validators: List[SelectCompatiableValidators] @staticmethod def split_str(cls, data: Any) -> Any: @@ -694,28 +749,33 @@ def py_type_workflow_step(self) -> Type: return optional(self.py_type_if_required()) def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation: + validators = {} + requires_value = self.request_requires_value + py_type = None if state_representation == "workflow_step": - return dynamic_model_information_from_py_type(self, self.py_type_workflow_step, requires_value=False) + py_type = self.py_type_workflow_step elif state_representation == "workflow_step_linked": py_type = self.py_type_if_required(allow_connections=True) - return dynamic_model_information_from_py_type( - self, optional_if_needed(py_type, self.optional or self.multiple) - ) + py_type = optional_if_needed(py_type, self.optional or self.multiple) elif state_representation == "test_case_xml": # in a YAML test case representation this can be string, in XML we are still expecting a comma separated string py_type = self.py_type_if_required(allow_connections=False) if self.multiple: validators = {"from_string": field_validator(self.name, mode="before")(SelectParameterModel.split_str)} - else: - validators = {} - return dynamic_model_information_from_py_type( - self, optional_if_needed(py_type, self.optional), validators=validators - ) + py_type = optional_if_needed(py_type, self.optional) + elif state_representation == "job_internal": + requires_value = True + py_type = self.py_type else: - requires_value = self.request_requires_value - if state_representation == "job_internal": - requires_value = True - return dynamic_model_information_from_py_type(self, self.py_type, requires_value=requires_value) + py_type = self.py_type + + validator_models = static_tool_validators_to_pydantic(self.validators) + if validator_models: + py_type = expand_annotation(py_type, validator_models) + + return dynamic_model_information_from_py_type( + self, py_type, validators=validators, requires_value=requires_value + ) @property def has_selected_static_option(self): diff --git a/lib/galaxy/tool_util/parser/interface.py b/lib/galaxy/tool_util/parser/interface.py index 238f6d2b9268..c1a23f410927 100644 --- a/lib/galaxy/tool_util/parser/interface.py +++ b/lib/galaxy/tool_util/parser/interface.py @@ -28,6 +28,7 @@ from galaxy.util import Element from galaxy.util.path import safe_walk +from .parameter_validators import AnyValidatorModel from .util import _parse_name if TYPE_CHECKING: @@ -502,7 +503,7 @@ def parse_sanitizer_elem(self): """ return None - def parse_validator_elems(self): + def parse_validators(self) -> List[AnyValidatorModel]: """Return an XML description of sanitizers. This is a stop gap until we can rework galaxy.tools.parameters.validation to not explicitly depend on XML. diff --git a/lib/galaxy/tool_util/parser/parameter_validators.py b/lib/galaxy/tool_util/parser/parameter_validators.py new file mode 100644 index 000000000000..8b363e5e53cd --- /dev/null +++ b/lib/galaxy/tool_util/parser/parameter_validators.py @@ -0,0 +1,699 @@ +import json +import os.path +from typing import ( + Any, + cast, + List, + Optional, + Union, +) + +import regex +from pydantic import ( + BaseModel, + ConfigDict, + Field, + model_validator, + PrivateAttr, +) +from typing_extensions import ( + Annotated, + get_args, + Literal, + Protocol, +) + +from galaxy.util import ( + asbool, + Element, +) + + +class ValidationArgument: + doc: Optional[str] + xml_body: bool + xml_allow_json_load: bool + + def __init__( + self, + doc: Optional[str], + xml_body: bool = False, + xml_allow_json_load: bool = False, + ): + self.doc = doc + self.xml_body = xml_body + self.xml_allow_json_load = xml_allow_json_load + + +Negate = Annotated[ + bool, + ValidationArgument("Negates the result of the validator."), +] +NEGATE_DEFAULT = False +SPLIT_DEFAULT = "\t" +DEFAULT_VALIDATOR_MESSAGE = "Parameter validation error." + +ValidatorType = Literal[ + "expression", + "regex", + "in_range", + "length", + "metadata", + "dataset_metadata_equal", + "unspecified_build", + "no_options", + "empty_field", + "empty_dataset", + "empty_extra_files_path", + "dataset_metadata_in_data_table", + "dataset_metadata_not_in_data_table", + "dataset_metadata_in_range", + "value_in_data_table", + "value_not_in_data_table", + "dataset_ok_validator", + "dataset_metadata_in_file", +] + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class ParameterValidatorModel(StrictModel): + type: ValidatorType + message: Annotated[ + Optional[str], + ValidationArgument( + """The error message displayed on the tool form if validation fails. A placeholder string ``%s`` will be repaced by the ``value``""" + ), + ] = None + _static: bool = PrivateAttr(False) + _deprecated: bool = PrivateAttr(False) + + @model_validator(mode="after") + def set_default_message(cls, parameter_model: "ParameterValidatorModel"): + if parameter_model.message is None: + parameter_model.message = parameter_model.default_message + return parameter_model + + @property + def default_message(self) -> str: + return DEFAULT_VALIDATOR_MESSAGE + + +class StaticValidatorModel(ParameterValidatorModel): + _static: bool = PrivateAttr(True) + + def statically_validate(self, v: Any) -> None: ... + + +class ExpressionParameterValidatorModel(StaticValidatorModel): + """Check if a one line python expression given expression evaluates to True. + + The expression is given is the content of the validator tag.""" + + type: Literal["expression"] = "expression" + negate: Negate = NEGATE_DEFAULT + expression: Annotated[str, ValidationArgument("Python expression to validate.", xml_body=True)] + + def statically_validate(self, value: Any) -> None: + ExpressionParameterValidatorModel.expression_validation(self.expression, value, self) + + @staticmethod + def ensure_compiled(expression: Union[str, Any]) -> Any: + if isinstance(expression, str): + return compile(expression, "", "eval") + else: + return expression + + @staticmethod + def expression_validation( + expression: str, value: Any, validator: "ValidatorDescription", compiled_expression: Optional[Any] = None + ): + if compiled_expression is None: + compiled_expression = ExpressionParameterValidatorModel.ensure_compiled(expression) + message = None + try: + evalresult = eval(compiled_expression, dict(value=value)) + except Exception: + message = f"Validator '{expression}' could not be evaluated on '{value}'" + evalresult = False + + raise_error_if_valiation_fails(bool(evalresult), validator, message=message, value_to_show=value) + + @property + def default_message(self) -> str: + return f"Value '%s' does not evaluate to {'True' if not self.negate else 'False'} for '{self.expression}'" + + +class RegexParameterValidatorModel(StaticValidatorModel): + """Check if a regular expression **matches** the value, i.e. appears + at the beginning of the value. To enforce a match of the complete value use + ``$`` at the end of the expression. The expression is given is the content + of the validator tag. Note that for ``selects`` each option is checked + separately.""" + + type: Literal["regex"] = "regex" + negate: Negate = NEGATE_DEFAULT + expression: Annotated[str, ValidationArgument("Regular expression to validate against.", xml_body=True)] + + @property + def default_message(self) -> str: + return f"Value '%s' does {'not ' if not self.negate else ''}match regular expression '{self.expression.replace('%', '%%')}'" + + def statically_validate(self, value: Any) -> None: + RegexParameterValidatorModel.regex_validation(self.expression, value, self) + + @staticmethod + def regex_validation(expression: str, value: Any, validator: "ValidatorDescription"): + if not isinstance(value, list): + value = [value] + for val in value: + match = regex.match(expression, val or "") + raise_error_if_valiation_fails(match is not None, validator, value_to_show=val) + + +class InRangeParameterValidatorModel(StaticValidatorModel): + type: Literal["in_range"] = "in_range" + min: Optional[Union[float, int]] = None + max: Optional[Union[float, int]] = None + exclude_min: bool = False + exclude_max: bool = False + negate: Negate = NEGATE_DEFAULT + + def statically_validate(self, value: Any): + if isinstance(value, (int, float)): + validates = True + if self.min is not None and value == self.min and self.exclude_min: + validates = False + elif self.min is not None and value < self.min: + validates = False + elif self.max is not None and value == self.max and self.exclude_max: + validates = False + if self.max is not None and value > self.max: + validates = False + raise_error_if_valiation_fails(validates, self) + + @property + def default_message(self) -> str: + op1 = "<=" + op2 = "<=" + if self.exclude_min: + op1 = "<" + if self.exclude_max: + op2 = "<" + range_description_str = f"({self.min} {op1} value {op2} {self.max})" + return f"Value ('%s') must {'not ' if self.negate else ''}fulfill {range_description_str}" + + +class LengthParameterValidatorModel(StaticValidatorModel): + type: Literal["length"] = "length" + min: Optional[int] = None + max: Optional[int] = None + negate: Negate = NEGATE_DEFAULT + + def statically_validate(self, value: Any): + if isinstance(value, str): + length = len(value) + validates = True + if self.min is not None and length < self.min: + validates = False + if self.max is not None and length > self.max: + validates = False + raise_error_if_valiation_fails(validates, self) + + @property + def default_message(self) -> str: + return f"Must {'not ' if self.negate else ''}have length of at least {self.min} and at most {self.max}" + + +class MetadataParameterValidatorModel(ParameterValidatorModel): + type: Literal["metadata"] = "metadata" + check: Optional[List[str]] = None + skip: Optional[List[str]] = None + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + check = self.check + skip = self.skip + message = DEFAULT_VALIDATOR_MESSAGE + if not self.negate: + message = "Metadata '%s' missing, click the pencil icon in the history item to edit / save the metadata attributes" + else: + if check: + message = f"""At least one of the checked metadata '{",".join(check)}' is set, click the pencil icon in the history item to edit / save the metadata attributes""" + elif skip: + message = f"""At least one of the non skipped metadata '{",".join(skip)}' is set, click the pencil icon in the history item to edit / save the metadata attributes""" + return message + + +class DatasetMetadataEqualParameterValidatorModel(ParameterValidatorModel): + type: Literal["dataset_metadata_equal"] = "dataset_metadata_equal" + metadata_name: str + value: Annotated[Any, ValidationArgument("Value to test against", xml_allow_json_load=True)] + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + if not self.negate: + message = f"Metadata value for '{self.metadata_name}' must be '{self.value}', but it is '%s'." + else: + message = f"Metadata value for '{self.metadata_name}' must not be '{self.value}' but it is." + + +class UnspecifiedBuildParameterValidatorModel(ParameterValidatorModel): + type: Literal["unspecified_build"] = "unspecified_build" + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + return f"{'Unspecified' if not self.negate else 'Specified'} genome build, click the pencil icon in the history item to {'set' if not self.negate else 'remove'} the genome build" + + +class NoOptionsParameterValidatorModel(StaticValidatorModel): + type: Literal["no_options"] = "no_options" + negate: Negate = NEGATE_DEFAULT + + @staticmethod + def no_options_validate(value: Any, validator: "ValidatorDescription"): + raise_error_if_valiation_fails(value is not None, validator) + + def statically_validate(self, value: Any) -> None: + NoOptionsParameterValidatorModel.no_options_validate(value, self) + + @property + def default_message(self) -> str: + return f"{'No options' if self.negate == 'false' else 'Options'} available for selection" + + +class EmptyFieldParameterValidatorModel(StaticValidatorModel): + type: Literal["empty_field"] = "empty_field" + negate: Negate = NEGATE_DEFAULT + + @staticmethod + def empty_validate(value: Any, validator: "ValidatorDescription"): + raise_error_if_valiation_fails((value != ""), validator) + + def statically_validate(self, value: Any) -> None: + EmptyFieldParameterValidatorModel.empty_validate(value, self) + + @property + def default_message(self) -> str: + if not self.negate: + message = "Field requires a value" + else: + message = "Field must not set a value" + return message + + +class EmptyDatasetParameterValidatorModel(ParameterValidatorModel): + type: Literal["empty_dataset"] = "empty_dataset" + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + return f"The selected dataset is {'non-' if self.negate else ''}empty, this tool expects {'non-' if not self.negate else ''}empty files." + + +class EmptyExtraFilesPathParameterValidatorModel(ParameterValidatorModel): + type: Literal["empty_extra_files_path"] = "empty_extra_files_path" + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + negate = self.negate + return f"The selected dataset's extra_files_path directory is {'non-' if negate else ''}empty or does {'not ' if not negate else ''}exist, this tool expects {'non-' if not negate else ''}empty extra_files_path directories associated with the selected input." + + +class DatasetMetadataInDataTableParameterValidatorModel(ParameterValidatorModel): + type: Literal["dataset_metadata_in_data_table"] = "dataset_metadata_in_data_table" + table_name: str + metadata_name: str + metadata_column: int + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + return f"Value for metadata {self.metadata_name} was not found in {self.table_name}." + + +class DatasetMetadataNotInDataTableParameterValidatorModel(ParameterValidatorModel): + type: Literal["dataset_metadata_not_in_data_table"] = "dataset_metadata_not_in_data_table" + table_name: str + metadata_name: str + metadata_column: int + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + return f"Value for metadata {self.metadata_name} was not found in {self.table_name}." + + +class DatasetMetadataInRangeParameterValidatorModel(ParameterValidatorModel): + type: Literal["dataset_metadata_in_range"] = "dataset_metadata_in_range" + metadata_name: str + min: Optional[Union[float, int]] = None + max: Optional[Union[float, int]] = None + exclude_min: bool = False + exclude_max: bool = False + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + op1 = "<=" + op2 = "<=" + if self.exclude_min: + op1 = "<" + if self.exclude_max: + op2 = "<" + range_description_str = f"({self.min} {op1} value {op2} {self.max})" + return f"Value ('%s') must {'not ' if self.negate else ''}fulfill {range_description_str}" + + +class ValueInDataTableParameterValidatorModel(ParameterValidatorModel): + type: Literal["value_in_data_table"] = "value_in_data_table" + table_name: str + metadata_column: int + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + return "Value for metadata not found." + + +class ValueNotInDataTableParameterValidatorModel(ParameterValidatorModel): + type: Literal["value_not_in_data_table"] = "value_not_in_data_table" + table_name: str + metadata_column: int + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + return f"Value was not found in {self.table_name}." + + +class DatasetOkValidatorParameterValidatorModel(ParameterValidatorModel): + type: Literal["dataset_ok_validator"] = "dataset_ok_validator" + negate: Negate = NEGATE_DEFAULT + + @property + def default_message(self) -> str: + if not self.negate: + message = ( + "The selected dataset is still being generated, select another dataset or wait until it is completed" + ) + else: + message = "The selected dataset must not be in state OK" + return message + + +class DatasetMetadataInFileParameterValidatorModel(ParameterValidatorModel): + type: Literal["dataset_metadata_in_file"] = "dataset_metadata_in_file" + filename: str + metadata_name: str + metadata_column: int + line_startswith: Optional[str] = None + split: str = SPLIT_DEFAULT + negate: Negate = NEGATE_DEFAULT + _deprecated: bool = PrivateAttr(True) + + @property + def default_message(self) -> str: + return f"Value for metadata {self.metadata_name} was not found in {self.filename}." + + +AnyValidatorModel = Annotated[ + Union[ + ExpressionParameterValidatorModel, + RegexParameterValidatorModel, + InRangeParameterValidatorModel, + LengthParameterValidatorModel, + MetadataParameterValidatorModel, + DatasetMetadataEqualParameterValidatorModel, + UnspecifiedBuildParameterValidatorModel, + NoOptionsParameterValidatorModel, + EmptyFieldParameterValidatorModel, + EmptyDatasetParameterValidatorModel, + EmptyExtraFilesPathParameterValidatorModel, + DatasetMetadataInDataTableParameterValidatorModel, + DatasetMetadataNotInDataTableParameterValidatorModel, + DatasetMetadataInRangeParameterValidatorModel, + ValueInDataTableParameterValidatorModel, + ValueNotInDataTableParameterValidatorModel, + DatasetOkValidatorParameterValidatorModel, + DatasetMetadataInFileParameterValidatorModel, + ], + Field(discriminator="type"), +] + + +def parse_xml_validators(input_elem: Element) -> List[AnyValidatorModel]: + validator_els: List[Element] = input_elem.findall("validator") or [] + models = [] + for validator_el in validator_els: + models.append(parse_xml_validator(validator_el)) + return models + + +def static_validators(validator_models: List[AnyValidatorModel]) -> List[AnyValidatorModel]: + static_validators = [] + for validator_model in validator_models: + print(validator_model._static) + if validator_model._static: + static_validators.append(validator_model) + return static_validators + + +def parse_xml_validator(validator_el: Element) -> AnyValidatorModel: + _type = validator_el.get("type") + if _type is None: + raise ValueError("Required 'type' attribute missing from validator") + valid_types = get_args(ValidatorType) + if _type not in valid_types: + raise ValueError(f"Unknown 'type' attribute in validator {_type}") + validator_type: ValidatorType = cast(ValidatorType, _type) + if validator_type == "expression": + return ExpressionParameterValidatorModel( + type="expression", + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + expression=validator_el.text, + ) + elif validator_type == "regex": + return RegexParameterValidatorModel( + type="regex", + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + expression=validator_el.text, + ) + elif validator_type == "in_range": + return InRangeParameterValidatorModel( + type="in_range", + message=_parse_message(validator_el), + min=_parse_number(validator_el, "min"), + max=_parse_number(validator_el, "max"), + exclude_min=_parse_bool(validator_el, "exclude_min", False), + exclude_max=_parse_bool(validator_el, "exclude_max", False), + negate=_parse_negate(validator_el), + ) + elif validator_type == "length": + return LengthParameterValidatorModel( + type="length", + min=_parse_int(validator_el, "min"), + max=_parse_int(validator_el, "max"), + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "metadata": + return MetadataParameterValidatorModel( + type="metadata", + message=_parse_message(validator_el), + check=_parse_str_list(validator_el, "check"), + skip=_parse_str_list(validator_el, "skip"), + negate=_parse_negate(validator_el), + ) + elif validator_type == "dataset_metadata_equal": + return DatasetMetadataEqualParameterValidatorModel( + type="dataset_metadata_equal", + metadata_name=validator_el.get("metadata_name"), + value=_parse_json_value(validator_el), + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "unspecified_build": + return UnspecifiedBuildParameterValidatorModel( + type="unspecified_build", + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "no_options": + return NoOptionsParameterValidatorModel( + type="no_options", + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "empty_field": + return EmptyFieldParameterValidatorModel( + type="empty_field", + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "empty_dataset": + return EmptyDatasetParameterValidatorModel( + type="empty_dataset", + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "empty_extra_files_path": + return EmptyExtraFilesPathParameterValidatorModel( + type="empty_extra_files_path", + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "dataset_metadata_in_data_table": + return DatasetMetadataInDataTableParameterValidatorModel( + type="dataset_metadata_in_data_table", + message=_parse_message(validator_el), + table_name=validator_el.get("table_name"), + metadata_name=validator_el.get("metadata_name"), + metadata_column=_parse_metadata_column(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "dataset_metadata_not_in_data_table": + return DatasetMetadataNotInDataTableParameterValidatorModel( + type="dataset_metadata_not_in_data_table", + message=_parse_message(validator_el), + table_name=validator_el.get("table_name"), + metadata_name=validator_el.get("metadata_name"), + metadata_column=_parse_metadata_column(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "dataset_metadata_in_range": + return DatasetMetadataInRangeParameterValidatorModel( + type="dataset_metadata_in_range", + message=_parse_message(validator_el), + metadata_name=validator_el.get("metadata_name"), + min=_parse_number(validator_el, "min"), + max=_parse_number(validator_el, "max"), + exclude_min=_parse_bool(validator_el, "exclude_min", False), + exclude_max=_parse_bool(validator_el, "exclude_max", False), + negate=_parse_negate(validator_el), + ) + elif validator_type == "value_in_data_table": + return ValueInDataTableParameterValidatorModel( + type="value_in_data_table", + message=_parse_message(validator_el), + table_name=validator_el.get("table_name"), + metadata_column=_parse_metadata_column(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "value_not_in_data_table": + return ValueNotInDataTableParameterValidatorModel( + type="value_not_in_data_table", + message=_parse_message(validator_el), + table_name=validator_el.get("table_name"), + metadata_column=_parse_metadata_column(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "dataset_ok_validator": + return DatasetOkValidatorParameterValidatorModel( + type="dataset_ok_validator", + message=_parse_message(validator_el), + negate=_parse_negate(validator_el), + ) + elif validator_type == "dataset_metadata_in_file": + filename = validator_el.get("filename") + assert os.path.exists(filename), f"File {filename} specified by the 'filename' attribute not found" + return DatasetMetadataInFileParameterValidatorModel( + type="dataset_metadata_in_file", + message=_parse_message(validator_el), + filename=filename, + metadata_name=validator_el.get("metadata_name"), + metadata_column=_parse_metadata_column(validator_el), + line_startswith=validator_el.get("line_startswith"), + split=validator_el.get("split", SPLIT_DEFAULT), + negate=_parse_negate(validator_el), + ) + else: + raise ValueError(f"Unhandled 'type' attribute in validator {validator_type}") + + +class ValidatorDescription(Protocol): + + @property + def negate(self) -> bool: ... + + @property + def message(self) -> Optional[str]: ... + + +def raise_error_if_valiation_fails( + value: bool, validator: ValidatorDescription, message: Optional[str] = None, value_to_show: Optional[str] = None +): + if not isinstance(value, bool): + raise AssertionError("Validator logic problem - computed validation value must be boolean") + if message is None: + message = validator.message + if message is None: + message = DEFAULT_VALIDATOR_MESSAGE + assert message + if value_to_show and "%s" in message: + message = message % value_to_show + negate = validator.negate + if (not negate and value) or (negate and not value): + return + else: + raise ValueError(message) + + +def _parse_message(xml_el: Element) -> Optional[str]: + message = cast(Optional[str], xml_el.get("message")) + return message + + +def _parse_int(xml_el: Element, attribute: str) -> Optional[int]: + raw_value = xml_el.get(attribute) + if raw_value: + return int(raw_value) + else: + return None + + +def _parse_number(xml_el: Element, attribute: str) -> Optional[Union[float, int]]: + raw_value = xml_el.get(attribute) + if raw_value and ("." in raw_value or "e" in raw_value): + return float(raw_value) + elif raw_value: + return int(raw_value) + else: + return None + + +def _parse_negate(xml_el: Element) -> bool: + return _parse_bool(xml_el, "negate", False) + + +def _parse_bool(xml_el: Element, attribute: str, default_value: bool) -> bool: + return asbool(xml_el.get(attribute, default_value)) + + +def _parse_str_list(xml_el: Element, attribute: str) -> List[str]: + raw_value = xml_el.get(attribute) + if not raw_value: + return [] + else: + return [v.strip() for v in raw_value.split(",")] + + +def _parse_json_value(xml_el: Element) -> Any: + value = xml_el.get("value", None) or json.loads(xml_el.get("value_json", "null")) + return value + + +def _parse_metadata_column(xml_el: Element) -> Union[int, str]: + column = xml_el.get("metadata_column", 0) + try: + return int(column) + except ValueError: + return column diff --git a/lib/galaxy/tool_util/parser/xml.py b/lib/galaxy/tool_util/parser/xml.py index b94cf7a84457..06a25c78f29b 100644 --- a/lib/galaxy/tool_util/parser/xml.py +++ b/lib/galaxy/tool_util/parser/xml.py @@ -71,6 +71,10 @@ ToolOutputCollection, ToolOutputCollectionStructure, ) +from .parameter_validators import ( + AnyValidatorModel, + parse_xml_validators, +) from .stdio import ( aggressive_error_checks, error_on_exit_code, @@ -1339,8 +1343,8 @@ def parse_help(self): def parse_sanitizer_elem(self): return self.input_elem.find("sanitizer") - def parse_validator_elems(self): - return self.input_elem.findall("validator") + def parse_validators(self) -> List[AnyValidatorModel]: + return parse_xml_validators(self.input_elem) def parse_dynamic_options(self) -> Optional[XmlDynamicOptions]: """Return a XmlDynamicOptions to describe dynamic options if options elem is available.""" diff --git a/lib/galaxy/tool_util/unittest_utils/sample_data.py b/lib/galaxy/tool_util/unittest_utils/sample_data.py index d4b6ddb6f027..e9b19283401a 100644 --- a/lib/galaxy/tool_util/unittest_utils/sample_data.py +++ b/lib/galaxy/tool_util/unittest_utils/sample_data.py @@ -17,3 +17,56 @@ """ ) + +VALID_XML_VALIDATORS = [ + """""", + """""", + """""", + """value == 7""", + """mycoolexpression""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", +] + +INVALID_XML_VALIDATORS = [ + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""", + """""" + """""" + """""", + """""", + """""", + """""", +] diff --git a/lib/galaxy/tools/parameters/basic.py b/lib/galaxy/tools/parameters/basic.py index 31e54d4a160c..8a26ef1c8b96 100644 --- a/lib/galaxy/tools/parameters/basic.py +++ b/lib/galaxy/tools/parameters/basic.py @@ -196,9 +196,7 @@ def __init__(self, tool, input_source, context=None): self.sanitizer = ToolParameterSanitizer.from_element(sanitizer_elem) else: self.sanitizer = None - self.validators = [] - for elem in input_source.parse_validator_elems(): - self.validators.append(validation.Validator.from_element(self, elem)) + self.validators = validation.to_validators(tool.app, input_source.parse_validators()) @property def visible(self) -> bool: @@ -2046,7 +2044,7 @@ def __init__(self, tool, input_source, trans=None): self.load_contents = int(input_source.get("load_contents", 0)) # Add metadata validator if not input_source.get_bool("no_validation", False): - self.validators.append(validation.MetadataValidator()) + self.validators.append(validation.MetadataValidator.default_metadata_validator()) self._parse_formats(trans, input_source) tag = input_source.get("tag") self.multiple = input_source.get_bool("multiple", False) @@ -2467,7 +2465,10 @@ def from_json(self, value, trans, other_values=None): rval = value elif isinstance(value, MutableMapping) and "src" in value and "id" in value: if value["src"] == "hdca": - rval = cast(HistoryDatasetCollectionAssociation, src_id_to_item(sa_session=trans.sa_session, value=value, security=trans.security)) + rval = cast( + HistoryDatasetCollectionAssociation, + src_id_to_item(sa_session=trans.sa_session, value=value, security=trans.security), + ) elif isinstance(value, list): if len(value) > 0: value = value[0] diff --git a/lib/galaxy/tools/parameters/dynamic_options.py b/lib/galaxy/tools/parameters/dynamic_options.py index e354c93eabb5..8d9488128a03 100644 --- a/lib/galaxy/tools/parameters/dynamic_options.py +++ b/lib/galaxy/tools/parameters/dynamic_options.py @@ -621,8 +621,9 @@ def load_from_parameter(from_parameter, transform_lines=None): self.filters.append(Filter.from_element(self, filter_elem)) # Load Validators - for validator in elem.findall("validator"): - self.validators.append(validation.Validator.from_element(self.tool_param, validator)) + validators = validation.parse_xml_validators(self.tool_param.tool.app, elem) + if validators: + self.validators = validators if self.dataset_ref_name: tool_param.data_ref = self.dataset_ref_name diff --git a/lib/galaxy/tools/parameters/validation.py b/lib/galaxy/tools/parameters/validation.py index 6334fd95f8b8..8c40e2278500 100644 --- a/lib/galaxy/tools/parameters/validation.py +++ b/lib/galaxy/tools/parameters/validation.py @@ -3,16 +3,27 @@ """ import abc -import json import logging -import os.path - -import regex +from typing import ( + Any, + List, + Optional, + Union, +) from galaxy import ( model, util, ) +from galaxy.tool_util.parser.parameter_validators import ( + AnyValidatorModel, + EmptyFieldParameterValidatorModel, + ExpressionParameterValidatorModel, + MetadataParameterValidatorModel, + parse_xml_validators as parse_xml_validators_models, + raise_error_if_valiation_fails, + RegexParameterValidatorModel, +) log = logging.getLogger(__name__) @@ -24,27 +35,7 @@ class Validator(abc.ABC): requires_dataset_metadata = False - @classmethod - def from_element(cls, param, elem): - """ - Initialize the appropriate Validator class - - example call `validation.Validator.from_element(ToolParameter_object, Validator_object)` - - needs to be implemented in the subclasses and should return the - corresponding Validator object by a call to `cls( ... )` which calls the - `__init__` method of the corresponding validator - - param cls the Validator class - param param the element to be evaluated (which contains the validator) - param elem the validator element - return an object of a Validator subclass that corresponds to the type attribute of the validator element - """ - _type = elem.get("type") - assert _type is not None, "Required 'type' attribute missing from validator" - return validator_types[_type].from_element(param, elem) - - def __init__(self, message, negate=False): + def __init__(self, message: str, negate: bool = False): self.message = message self.negate = util.asbool(negate) super().__init__() @@ -68,15 +59,7 @@ def validate(self, value, trans=None, message=None, value_to_show=None): return None if positive validation, otherwise a ValueError is raised """ - assert isinstance(value, bool), "value must be boolean" - if message is None: - message = self.message - if value_to_show and "%s" in message: - message = message % value_to_show - if (not self.negate and value) or (self.negate and not value): - return - else: - raise ValueError(message) + raise_error_if_valiation_fails(value, self, message=message, value_to_show=value_to_show) class RegexValidator(Validator): @@ -84,24 +67,14 @@ class RegexValidator(Validator): Validator that evaluates a regular expression """ - @classmethod - def from_element(cls, param, elem): - return cls(elem.get("message"), elem.text, elem.get("negate", "false")) - - def __init__(self, message, expression, negate): - if message is None: - message = f"Value '%s' does {'not ' if negate == 'false' else ''}match regular expression '{expression.replace('%', '%%')}'" + def __init__(self, message: str, expression: str, negate: bool): super().__init__(message, negate) # Compile later. RE objects used to not be thread safe. Not sure about # the sre module. self.expression = expression def validate(self, value, trans=None): - if not isinstance(value, list): - value = [value] - for val in value: - match = regex.match(self.expression, val or "") - super().validate(match is not None, value_to_show=val) + RegexParameterValidatorModel.regex_validation(self.expression, value, self) class ExpressionValidator(Validator): @@ -109,24 +82,16 @@ class ExpressionValidator(Validator): Validator that evaluates a python expression using the value """ - @classmethod - def from_element(cls, param, elem): - return cls(elem.get("message"), elem.text, elem.get("negate", "false")) - - def __init__(self, message, expression, negate): - if message is None: - message = f"Value '%s' does not evaluate to {'True' if negate == 'false' else 'False'} for '{expression}'" + def __init__(self, message: str, expression: str, negate: bool): super().__init__(message, negate) self.expression = expression # Save compiled expression, code objects are thread safe (right?) - self.compiled_expression = compile(expression, "", "eval") + self.compiled_expression = ExpressionParameterValidatorModel.ensure_compiled(expression) def validate(self, value, trans=None): - try: - evalresult = eval(self.compiled_expression, dict(value=value)) - except Exception: - super().validate(False, message=f"Validator '{self.expression}' could not be evaluated on '{value}'") - super().validate(bool(evalresult), value_to_show=value) + ExpressionParameterValidatorModel.expression_validation( + self.expression, value, self, compiled_expression=self.compiled_expression + ) class InRangeValidator(ExpressionValidator): @@ -134,18 +99,15 @@ class InRangeValidator(ExpressionValidator): Validator that ensures a number is in a specified range """ - @classmethod - def from_element(cls, param, elem): - return cls( - elem.get("message"), - elem.get("min"), - elem.get("max"), - elem.get("exclude_min", "false"), - elem.get("exclude_max", "false"), - elem.get("negate", "false"), - ) - - def __init__(self, message, range_min, range_max, exclude_min=False, exclude_max=False, negate=False): + def __init__( + self, + message: str = None, + min: Optional[float] = None, + max: Optional[float] = None, + exclude_min: bool = False, + exclude_max: bool = False, + negate: bool = False, + ): """ When the optional exclude_min and exclude_max attributes are set to true, the range excludes the end points (i.e., min < value < max), @@ -153,10 +115,10 @@ def __init__(self, message, range_min, range_max, exclude_min=False, exclude_max (1.e., min <= value <= max). Combinations of exclude_min and exclude_max values are allowed. """ - self.min = range_min if range_min is not None else "-inf" - self.exclude_min = util.asbool(exclude_min) - self.max = range_max if range_max is not None else "inf" - self.exclude_max = util.asbool(exclude_max) + self.min = str(min) if min is not None else "-inf" + self.exclude_min = exclude_min + self.max = str(max) if max is not None else "inf" + self.exclude_max = exclude_max assert float(self.min) <= float(self.max), "min must be less than or equal to max" # Remove unneeded 0s and decimal from floats to make message pretty. op1 = "<=" @@ -166,8 +128,6 @@ def __init__(self, message, range_min, range_max, exclude_min=False, exclude_max if self.exclude_max: op2 = "<" expression = f"float('{self.min}') {op1} float(value) {op2} float('{self.max}')" - if message is None: - message = f"Value ('%s') must {'not ' if negate == 'true' else ''}fulfill {expression}" super().__init__(message, expression, negate) @@ -176,14 +136,8 @@ class LengthValidator(InRangeValidator): Validator that ensures the length of the provided string (value) is in a specific range """ - @classmethod - def from_element(cls, param, elem): - return cls(elem.get("message"), elem.get("min"), elem.get("max"), elem.get("negate", "false")) - - def __init__(self, message, length_min, length_max, negate): - if message is None: - message = f"Must {'not ' if negate == 'true' else ''}have length of at least {length_min} and at most {length_max}" - super().__init__(message, range_min=length_min, range_max=length_max, negate=negate) + def __init__(self, message: str, min: float, max: float, negate: bool): + super().__init__(message, min=min, max=max, negate=negate) def validate(self, value, trans=None): if value is None: @@ -196,16 +150,8 @@ class DatasetOkValidator(Validator): Validator that checks if a dataset is in an 'ok' state """ - @classmethod - def from_element(cls, param, elem): - negate = elem.get("negate", "false") - message = elem.get("message") - if message is None: - if negate == "false": - message = "The selected dataset is still being generated, select another dataset or wait until it is completed" - else: - message = "The selected dataset must not be in state OK" - return cls(message, negate) + def __init__(self, message: str, negate: bool = False): + super().__init__(message, negate=negate) def validate(self, value, trans=None): if value: @@ -217,13 +163,8 @@ class DatasetEmptyValidator(Validator): Validator that checks if a dataset has a positive file size. """ - @classmethod - def from_element(cls, param, elem): - message = elem.get("message") - negate = elem.get("negate", "false") - if not message: - message = f"The selected dataset is {'non-' if negate == 'true' else ''}empty, this tool expects {'non-' if negate == 'false' else ''}empty files." - return cls(message, negate) + def __init__(self, message: str, negate: bool = False): + super().__init__(message, negate=negate) def validate(self, value, trans=None): if value: @@ -235,13 +176,8 @@ class DatasetExtraFilesPathEmptyValidator(Validator): Validator that checks if a dataset's extra_files_path exists and is not empty. """ - @classmethod - def from_element(cls, param, elem): - message = elem.get("message") - negate = elem.get("negate", "false") - if not message: - message = f"The selected dataset's extra_files_path directory is {'non-' if negate == 'true' else ''}empty or does {'not ' if negate == 'false' else ''}exist, this tool expects {'non-' if negate == 'false' else ''}empty extra_files_path directories associated with the selected input." - return cls(message, negate) + def __init__(self, message: str, negate: bool = False): + super().__init__(message, negate=negate) def validate(self, value, trans=None): if value: @@ -255,25 +191,20 @@ class MetadataValidator(Validator): requires_dataset_metadata = True - @classmethod - def from_element(cls, param, elem): - message = elem.get("message") - return cls( - message=message, check=elem.get("check", ""), skip=elem.get("skip", ""), negate=elem.get("negate", "false") - ) - - def __init__(self, message=None, check="", skip="", negate="false"): - if not message: - if not util.asbool(negate): - message = "Metadata '%s' missing, click the pencil icon in the history item to edit / save the metadata attributes" - else: - if check != "": - message = f"At least one of the checked metadata '{check}' is set, click the pencil icon in the history item to edit / save the metadata attributes" - elif skip != "": - message = f"At least one of the non skipped metadata '{skip}' is set, click the pencil icon in the history item to edit / save the metadata attributes" + def __init__( + self, + message: str, + check: Optional[List[str]] = None, + skip: Optional[List[str]] = None, + negate: bool = False, + ): super().__init__(message, negate) - self.check = check.split(",") if check else None - self.skip = skip.split(",") if skip else None + self.check = check + self.skip = skip + + @staticmethod + def default_metadata_validator() -> "MetadataValidator": + return _to_validator(None, MetadataParameterValidatorModel()) def validate(self, value, trans=None): if value: @@ -293,25 +224,10 @@ class MetadataEqualValidator(Validator): requires_dataset_metadata = True def __init__(self, metadata_name=None, value=None, message=None, negate="false"): - if not message: - if not util.asbool(negate): - message = f"Metadata value for '{metadata_name}' must be '{value}', but it is '%s'." - else: - message = f"Metadata value for '{metadata_name}' must not be '{value}' but it is." super().__init__(message, negate) self.metadata_name = metadata_name self.value = value - @classmethod - def from_element(cls, param, elem): - value = elem.get("value", None) or json.loads(elem.get("value_json", "null")) - return cls( - metadata_name=elem.get("metadata_name", None), - value=value, - message=elem.get("message", None), - negate=elem.get("negate", "false"), - ) - def validate(self, value, trans=None): if value: metadata_value = getattr(value.metadata, self.metadata_name) @@ -325,13 +241,8 @@ class UnspecifiedBuildValidator(Validator): requires_dataset_metadata = True - @classmethod - def from_element(cls, param, elem): - message = elem.get("message") - negate = elem.get("negate", "false") - if not message: - message = f"{'Unspecified' if negate == 'false' else 'Specified'} genome build, click the pencil icon in the history item to {'set' if negate == 'false' else 'remove'} the genome build" - return cls(message, negate) + def __init__(self, message: str, negate: bool = False): + super().__init__(message, negate=negate) def validate(self, value, trans=None): # if value is None, we cannot validate @@ -348,13 +259,8 @@ class NoOptionsValidator(Validator): Validator that checks for empty select list """ - @classmethod - def from_element(cls, param, elem): - message = elem.get("message") - negate = elem.get("negate", "false") - if not message: - message = f"{'No options' if negate == 'false' else 'Options'} available for selection" - return cls(message, negate) + def __init__(self, message: str, negate: bool = False): + super().__init__(message, negate=negate) def validate(self, value, trans=None): super().validate(value is not None) @@ -365,19 +271,11 @@ class EmptyTextfieldValidator(Validator): Validator that checks for empty text field """ - @classmethod - def from_element(cls, param, elem): - message = elem.get("message") - negate = elem.get("negate", "false") - if not message: - if negate == "false": - message = elem.get("message", "Field requires a value") - else: - message = elem.get("message", "Field must not set a value") - return cls(message, negate) + def __init__(self, message: str, negate: bool = False): + super().__init__(message, negate=negate) def validate(self, value, trans=None): - super().validate(value != "") + EmptyFieldParameterValidatorModel.empty_validate(value, self) class MetadataInFileColumnValidator(Validator): @@ -391,33 +289,15 @@ class MetadataInFileColumnValidator(Validator): requires_dataset_metadata = True - @classmethod - def from_element(cls, param, elem): - filename = elem.get("filename") - assert filename, f"Required 'filename' attribute missing from {elem.get('type')} validator." - filename = f"{param.tool.app.config.tool_data_path}/{filename.strip()}" - assert os.path.exists(filename), f"File {filename} specified by the 'filename' attribute not found" - metadata_name = elem.get("metadata_name") - assert metadata_name, f"Required 'metadata_name' attribute missing from {elem.get('type')} validator." - metadata_name = metadata_name.strip() - metadata_column = int(elem.get("metadata_column", 0)) - split = elem.get("split", "\t") - message = elem.get("message", f"Value for metadata {metadata_name} was not found in {filename}.") - line_startswith = elem.get("line_startswith") - if line_startswith: - line_startswith = line_startswith.strip() - negate = elem.get("negate", "false") - return cls(filename, metadata_name, metadata_column, message, line_startswith, split, negate) - def __init__( self, - filename, - metadata_name, - metadata_column, - message="Value for metadata not found.", - line_startswith=None, - split="\t", - negate="false", + filename: str, + metadata_name: str, + metadata_column: int, + message: str = None, + line_startswith: Optional[str] = None, + split: str = "\t", + negate: bool = False, ): super().__init__(message, negate) self.metadata_name = metadata_name @@ -445,28 +325,20 @@ class ValueInDataTableColumnValidator(Validator): note: this is covered in a framework test (validation_value_in_datatable) """ - @classmethod - def from_element(cls, param, elem): - table_name = elem.get("table_name") - assert table_name, f"Required 'table_name' attribute missing from {elem.get('type')} validator." - tool_data_table = param.tool.app.tool_data_tables[table_name] - column = elem.get("metadata_column", 0) - try: - column = int(column) - except ValueError: - pass - message = elem.get("message", f"Value was not found in {table_name}.") - negate = elem.get("negate", "false") - return cls(tool_data_table, column, message, negate) - - def __init__(self, tool_data_table, column, message="Value not found.", negate="false"): + def __init__( + self, + tool_data_table, + metadata_column: Union[str, int], + message: str, + negate: bool = False, + ): super().__init__(message, negate) - self.valid_values = [] + self.valid_values: List[Any] = [] self._data_table_content_version = None self._tool_data_table = tool_data_table - if isinstance(column, str): - column = tool_data_table.columns[column] - self._column = column + if isinstance(metadata_column, str): + metadata_column = tool_data_table.columns[metadata_column] + self._column = metadata_column self._load_values() def _load_values(self): @@ -496,7 +368,9 @@ class ValueNotInDataTableColumnValidator(ValueInDataTableColumnValidator): note: this is covered in a framework test (validation_value_in_datatable) """ - def __init__(self, tool_data_table, metadata_column, message="Value already present.", negate="false"): + def __init__( + self, tool_data_table, metadata_column: Union[str, int], message="Value already present.", negate="false" + ): super().__init__(tool_data_table, metadata_column, message, negate) def validate(self, value, trans=None): @@ -517,26 +391,13 @@ class MetadataInDataTableColumnValidator(ValueInDataTableColumnValidator): requires_dataset_metadata = True - @classmethod - def from_element(cls, param, elem): - table_name = elem.get("table_name") - assert table_name, f"Required 'table_name' attribute missing from {elem.get('type')} validator." - tool_data_table = param.tool.app.tool_data_tables[table_name] - metadata_name = elem.get("metadata_name") - assert metadata_name, f"Required 'metadata_name' attribute missing from {elem.get('type')} validator." - metadata_name = metadata_name.strip() - # TODO rename to column? - metadata_column = elem.get("metadata_column", 0) - try: - metadata_column = int(metadata_column) - except ValueError: - pass - message = elem.get("message", f"Value for metadata {metadata_name} was not found in {table_name}.") - negate = elem.get("negate", "false") - return cls(tool_data_table, metadata_name, metadata_column, message, negate) - def __init__( - self, tool_data_table, metadata_name, metadata_column, message="Value for metadata not found.", negate="false" + self, + tool_data_table, + metadata_name: str, + metadata_column: Union[str, int], + message: str, + negate: bool = False, ): super().__init__(tool_data_table, metadata_column, message, negate) self.metadata_name = metadata_name @@ -558,7 +419,12 @@ class MetadataNotInDataTableColumnValidator(MetadataInDataTableColumnValidator): requires_dataset_metadata = True def __init__( - self, tool_data_table, metadata_name, metadata_column, message="Value for metadata not found.", negate="false" + self, + tool_data_table, + metadata_name: str, + metadata_column: Union[str, int], + message: str, + negate: bool = False, ): super().__init__(tool_data_table, metadata_name, metadata_column, message, negate) @@ -580,26 +446,18 @@ class MetadataInRangeValidator(InRangeValidator): requires_dataset_metadata = True - @classmethod - def from_element(cls, param, elem): - metadata_name = elem.get("metadata_name") - assert metadata_name, f"Required 'metadata_name' attribute missing from {elem.get('type')} validator." - metadata_name = metadata_name.strip() - ret = cls( - metadata_name, - elem.get("message"), - elem.get("min"), - elem.get("max"), - elem.get("exclude_min", "false"), - elem.get("exclude_max", "false"), - elem.get("negate", "false"), - ) - ret.message = "Metadata: " + ret.message - return ret - - def __init__(self, metadata_name, message, range_min, range_max, exclude_min, exclude_max, negate): + def __init__( + self, + metadata_name: str, + message: str = None, + min: Optional[float] = None, + max: Optional[float] = None, + exclude_min: bool = False, + exclude_max: bool = False, + negate: bool = False, + ): self.metadata_name = metadata_name - super().__init__(message, range_min, range_max, exclude_min, exclude_max, negate) + super().__init__(message, min, max, exclude_min, exclude_max, negate) def validate(self, value, trans=None): if value: @@ -638,3 +496,24 @@ def validate(self, value, trans=None): deprecated_validator_types = dict(dataset_metadata_in_file=MetadataInFileColumnValidator) validator_types.update(deprecated_validator_types) + + +def parse_xml_validators(app, xml_el: util.Element) -> List[Validator]: + return to_validators(app, parse_xml_validators_models(xml_el)) + + +def to_validators(app, validator_models: List[AnyValidatorModel]) -> Validator: + validators = [] + for validator_model in validator_models: + validators.append(_to_validator(app, validator_model)) + return validators + + +def _to_validator(app, validator_model: AnyValidatorModel) -> Validator: + as_dict = validator_model.model_dump() + validator_type = as_dict.pop("type") + if "table_name" in as_dict and app is not None: + table_name = as_dict.pop("table_name") + tool_data_table = app.tool_data_tables[table_name] + as_dict["tool_data_table"] = tool_data_table + return validator_types[validator_type](**as_dict) diff --git a/test/functional/tools/parameters/gx_int_validation_range.xml b/test/functional/tools/parameters/gx_int_validation_range.xml new file mode 100644 index 000000000000..5f7862497abf --- /dev/null +++ b/test/functional/tools/parameters/gx_int_validation_range.xml @@ -0,0 +1,15 @@ + + > '$output' + ]]> + + + + + + + + + + + diff --git a/test/functional/tools/parameters/gx_text_expression_validation.xml b/test/functional/tools/parameters/gx_text_expression_validation.xml new file mode 100644 index 000000000000..fe5113ec8fef --- /dev/null +++ b/test/functional/tools/parameters/gx_text_expression_validation.xml @@ -0,0 +1,20 @@ + + > '$output'; +cat '$inputs' >> $inputs_json; + ]]> + + + + + + 'foobar' in value + + + + + + + + + diff --git a/test/functional/tools/parameters/gx_text_length_validation.xml b/test/functional/tools/parameters/gx_text_length_validation.xml new file mode 100644 index 000000000000..4d5434aa6c58 --- /dev/null +++ b/test/functional/tools/parameters/gx_text_length_validation.xml @@ -0,0 +1,20 @@ + + > '$output'; +cat '$inputs' >> $inputs_json; + ]]> + + + + + + + + + + + + + + + diff --git a/test/functional/tools/parameters/gx_text_length_validation_negate.xml b/test/functional/tools/parameters/gx_text_length_validation_negate.xml new file mode 100644 index 000000000000..4b0b835fe562 --- /dev/null +++ b/test/functional/tools/parameters/gx_text_length_validation_negate.xml @@ -0,0 +1,20 @@ + + > '$output'; +cat '$inputs' >> $inputs_json; + ]]> + + + + + + + + + + + + + + + diff --git a/test/functional/tools/parameters/gx_text_regex_validation.xml b/test/functional/tools/parameters/gx_text_regex_validation.xml new file mode 100644 index 000000000000..2e1bd7325d8a --- /dev/null +++ b/test/functional/tools/parameters/gx_text_regex_validation.xml @@ -0,0 +1,20 @@ + + > '$output'; +cat '$inputs' >> $inputs_json; + ]]> + + + + + + ^[actg]*$ + + + + + + + + + diff --git a/test/unit/app/tools/test_parameter_validation.py b/test/unit/app/tools/test_parameter_validation.py index a7798927eb65..4b69034fac4b 100644 --- a/test/unit/app/tools/test_parameter_validation.py +++ b/test/unit/app/tools/test_parameter_validation.py @@ -220,12 +220,12 @@ def test_InRangeValidator(self): p.validate(10) with self.assertRaisesRegex( ValueError, - r"Parameter blah: Value \('15'\) must not fulfill float\('10'\) < float\(value\) <= float\('20'\)", + r"Parameter blah: Value \('15'\) must not fulfill \(10 < value <= 20\)", ): p.validate(15) with self.assertRaisesRegex( ValueError, - r"Parameter blah: Value \('20'\) must not fulfill float\('10'\) < float\(value\) <= float\('20'\)", + r"Parameter blah: Value \('20'\) must not fulfill \(10 < value <= 20\)", ): p.validate(20) p.validate(21) diff --git a/test/unit/app/tools/test_validation_parsing.py b/test/unit/app/tools/test_validation_parsing.py new file mode 100644 index 000000000000..c13b0973e6a0 --- /dev/null +++ b/test/unit/app/tools/test_validation_parsing.py @@ -0,0 +1,41 @@ +from typing import Optional + +from galaxy.tool_util.unittest_utils.sample_data import ( + INVALID_XML_VALIDATORS, + VALID_XML_VALIDATORS, +) +from galaxy.tools.parameters.validation import parse_xml_validators +from galaxy.util import XML + + +class MockApp: + + @property + def tool_data_tables(self): + return {"mycooltable": MockTable()} + + +class MockTable: + + def get_version_fields(self): + return (1, []) + + +def test_xml_validation_valid(): + for xml_validator in VALID_XML_VALIDATORS: + _validate_xml_str(xml_validator) + + +def test_xml_validation_invalid(): + for xml_validator in INVALID_XML_VALIDATORS: + exc: Optional[Exception] = None + try: + _validate_xml_str(xml_validator) + except ValueError as e: + exc = e + assert exc is not None, f"{xml_validator} - validated when it wasn't expected to" + + +def _validate_xml_str(xml_str: str): + xml_el = XML(f"{xml_str}") + parse_xml_validators(MockApp(), xml_el) diff --git a/test/unit/tool_util/parameter_specification.yml b/test/unit/tool_util/parameter_specification.yml index a46de9e0ec83..1c5249b0106e 100644 --- a/test/unit/tool_util/parameter_specification.yml +++ b/test/unit/tool_util/parameter_specification.yml @@ -191,6 +191,39 @@ gx_int_required: &gx_int_required gx_int_required_via_empty_string: <<: *gx_int_required + +gx_int_validation_range: + request_valid: + - parameter: 1 + - parameter: 9 + request_invalid: + - parameter: -1 + - parameter: 11 + - parameter: 10 + +gx_text_regex_validation: + request_valid: + - parameter: acgt + - parameter: a + - parameter: aaaggttac + request_invalid: + - parameter: acgu + - parameter: nucleic + +gx_text_expression_validation: + request_valid: + - parameter: the code was foobar + - parameter: foobar + request_invalid: + - parameter: the code was not foo bar + - parameter: '' + +gx_text_empty_validation: + request_valid: + - parameter: foobar + request_invalid: + - parameter: '' + gx_text: request_valid: &gx_text_request_valid - parameter: moocow @@ -294,6 +327,21 @@ gx_text_optional: - parameter: {} - parameter: { "moo": "cow" } +gx_text_length_validation: + request_valid: + - parameter: "mytext" + - parameter: "mytext123" + request_invalid: + - parameter: "s" # too short + - parameter: "mytext1231231231sd" # too long + +gx_text_length_validation_negate: + request_valid: + - parameter: "m" + - parameter: "mytext123mocowdowees" + request_invalid: + - parameter: "goldilocks" + gx_select: request_valid: - parameter: "--ex1" @@ -484,6 +532,12 @@ gx_select_multiple_optional: - parameter: {} - parameter: 5 +gx_select_no_options_validation: + job_internal_valid: + - parameter: "--ex1" + job_internal_invalid: + - {} + gx_genomebuild: request_valid: - parameter: hg19 diff --git a/test/unit/tool_util/test_parameter_validator_models.py b/test/unit/tool_util/test_parameter_validator_models.py new file mode 100644 index 000000000000..17af98f2ce6c --- /dev/null +++ b/test/unit/tool_util/test_parameter_validator_models.py @@ -0,0 +1,28 @@ +from typing import Optional + +from galaxy.tool_util.parser.parameter_validators import parse_xml_validators +from galaxy.tool_util.unittest_utils.sample_data import ( + INVALID_XML_VALIDATORS, + VALID_XML_VALIDATORS, +) +from galaxy.util import XML + + +def test_xml_validation_valid(): + for xml_validator in VALID_XML_VALIDATORS: + _validate_xml_str(xml_validator) + + +def test_xml_validation_invalid(): + for xml_validator in INVALID_XML_VALIDATORS: + exc: Optional[Exception] = None + try: + _validate_xml_str(xml_validator) + except ValueError as e: + exc = e + assert exc is not None, f"{xml_validator} - validated when it wasn't expected to" + + +def _validate_xml_str(xml_str: str): + xml_el = XML(f"{xml_str}") + parse_xml_validators(xml_el)