opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-01 15:37:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Pydantic models for parameter validators.

Browse Source
This commit is contained in:
John Chilton
2024-10-07 11:07:49 -04:00
parent c198c7cb97
commit 730ea2ee03
27 changed files with 1549 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/tool_util/parameters/_types.py
+11
View File
@@ -6,6 +6,7 @@ and build runtime models not to use mypy to type check static code.
"""
from typing import (
Any,
cast,
List,
Optional,
@@ -15,6 +16,7 @@ from typing import (
# 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]
lib/galaxy/tool_util/parameters/factory.py
+73 -3
View File
@@ -14,7 +14,19 @@ from galaxy.tool_util.parser.interface import (
PagesSource,
ToolSource,
)
from galaxy.tool_util.parser.util import parse_profile_version
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,
text_input_is_optional,
)
from galaxy.util import string_as_bool
from .models import (
BaseUrlParameterModel,
@@ -42,10 +54,13 @@ from .models import (
HiddenParameterModel,
IntegerParameterModel,
LabelValue,
NumberCompatiableValidators,
RepeatParameterModel,
RulesParameterModel,
SectionParameterModel,
SelectCompatiableValidators,
SelectParameterModel,
TextCompatiableValidators,
TextParameterModel,
ToolParameterBundle,
ToolParameterBundleModel,
@@ -82,7 +97,23 @@ 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())
int_validators: List[NumberCompatiableValidators] = []
for static_validator in static_validator_models:
if static_validator.type == "in_range":
int_validators.append(cast(InRangeParameterValidatorModel, static_validator))
min_raw = input_source.get("min", None)
max_raw = input_source.get("max", None)
min_int = int(min_raw) if min_raw is not None else None
max_int = int(max_raw) if max_raw is not None else None
return IntegerParameterModel(
name=input_source.parse_name(),
optional=optional,
value=int_value,
min=min_int,
max=max_int,
validators=int_validators,
)
elif param_type == "boolean":
nullable = input_source.parse_optional()
value = input_source.get_bool_or_none("checked", None if nullable else False)
@@ -92,10 +123,12 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
value=value,
)
elif param_type == "text":
optional = input_source.parse_optional()
optional, optionality_inferred = text_input_is_optional(input_source)
text_validators: List[TextCompatiableValidators] = _text_validators(input_source)
return TextParameterModel(
name=input_source.parse_name(),
optional=optional,
validators=text_validators,
)
elif param_type == "float":
optional = input_source.parse_optional()
@@ -107,18 +140,32 @@ 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())
float_validators: List[NumberCompatiableValidators] = []
for static_validator in static_validator_models:
if static_validator.type == "in_range":
float_validators.append(cast(InRangeParameterValidatorModel, static_validator))
min_raw = input_source.get("min", None)
max_raw = input_source.get("max", None)
min_float = float(min_raw) if min_raw is not None else None
max_float = float(max_raw) if max_raw is not None else None
return FloatParameterModel(
name=input_source.parse_name(),
optional=optional,
value=float_value,
min=min_float,
max=max_float,
validators=float_validators,
)
elif param_type == "hidden":
optional = input_source.parse_optional()
value = input_source.get("value")
hidden_validators: List[TextCompatiableValidators] = _text_validators(input_source)
return HiddenParameterModel(
name=input_source.parse_name(),
optional=optional,
value=value,
validators=hidden_validators,
)
elif param_type == "color":
optional = input_source.parse_optional()
@@ -158,11 +205,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())
select_validators: List[SelectCompatiableValidators] = []
for static_validator in static_validator_models:
if static_validator.type == "no_options":
select_validators.append(cast(NoOptionsParameterValidatorModel, static_validator))
return SelectParameterModel(
name=input_source.parse_name(),
optional=optional,
options=options,
multiple=multiple,
validators=select_validators,
)
elif param_type == "drill_down":
multiple = input_source.get_bool("multiple", False)
@@ -206,8 +259,10 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
multiple=multiple,
)
elif param_type == "directory_uri":
directory_uri_validators: List[TextCompatiableValidators] = _text_validators(input_source)
return DirectoryUriParameterModel(
name=input_source.parse_name(),
validators=directory_uri_validators,
)
else:
raise UnknownParameterTypeError(f"Unknown Galaxy parameter type {param_type}")
@@ -304,6 +359,21 @@ def _simple_cwl_type_to_model(simple_type: str, input_source: CwlInputSource):
)
def _text_validators(input_source: InputSource) -> List[TextCompatiableValidators]:
static_validator_models = static_validators(input_source.parse_validators())
text_validators: List[TextCompatiableValidators] = []
for static_validator in static_validator_models:
if static_validator.type == "length":
text_validators.append(cast(LengthParameterValidatorModel, static_validator))
elif static_validator.type == "regex":
text_validators.append(cast(RegexParameterValidatorModel, static_validator))
elif static_validator.type == "expression":
text_validators.append(cast(ExpressionParameterValidatorModel, static_validator))
elif static_validator.type == "empty_field":
text_validators.append(cast(EmptyFieldParameterValidatorModel, static_validator))
return text_validators
def _from_input_source_cwl(input_source: CwlInputSource) -> ToolParameterT:
schema_salad_field = input_source.field
if schema_salad_field is None:
lib/galaxy/tool_util/parameters/models.py
+100 -16
View File
@@ -11,11 +11,14 @@ from typing import (
Mapping,
NamedTuple,
Optional,
Sequence,
Type,
TypeVar,
Union,
)
from pydantic import (
AfterValidator,
AnyUrl,
BaseModel,
ConfigDict,
@@ -44,8 +47,18 @@ from galaxy.tool_util.parser.interface import (
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,
list_type,
optional,
@@ -54,7 +67,6 @@ from ._types import (
)
# TODO:
# - implement job vs request...
# - implement data_ref on rules and implement some cross model validation
# + request: Return info needed to build request pydantic model at runtime.
@@ -179,18 +191,64 @@ class LabelValue(BaseModel):
selected: bool
TextCompatiableValidators = Union[
LengthParameterValidatorModel,
RegexParameterValidatorModel,
ExpressionParameterValidatorModel,
EmptyFieldParameterValidatorModel,
]
def pydantic_to_galaxy_type(value: Any) -> Any:
"""We use advanced Pydantic types like URL but the Galaxy validators only expect strings for these."""
if isinstance(value, AnyUrl):
return str(value)
return value
VT = TypeVar("VT", bound=StaticValidatorModel)
def decorate_type_with_validators_if_needed(py_type: Type, static_validator_models: Sequence[VT]) -> Type:
pydantic_validator = pydantic_validator_for(static_validator_models)
if pydantic_validator:
return expand_annotation(py_type, [pydantic_validator])
else:
return py_type
# Looks like Annotated only work with one PlainValidator so condensing all static validators
# into a single PlainValidator for pydantic.
def pydantic_validator_for(static_validator_models: Sequence[VT]) -> Optional[AfterValidator]:
if static_validator_models:
def validator(v: Any) -> Any:
gx_val = pydantic_to_galaxy_type(v)
for static_validator_model in static_validator_models:
static_validator_model.statically_validate(gx_val)
return v
return AfterValidator(validator)
else:
return None
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:
return optional_if_needed(StrictStr, self.optional)
def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
py_type = self.py_type
py_type = decorate_type_with_validators_if_needed(self.py_type, self.validators)
if state_representation == "workflow_step_linked":
py_type = allow_connected_value(py_type)
requires_value = self.request_requires_value
@@ -203,12 +261,16 @@ class TextParameterModel(BaseGalaxyToolParameterModelDefinition):
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] = []
@property
def py_type(self) -> Type:
@@ -216,6 +278,10 @@ class IntegerParameterModel(BaseGalaxyToolParameterModelDefinition):
def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
py_type = self.py_type
validators = self.validators[:]
if self.min is not None or self.max is not None:
validators.append(InRangeParameterValidatorModel(min=self.min, max=self.max, implicit=True))
py_type = decorate_type_with_validators_if_needed(py_type, validators)
if state_representation == "workflow_step_linked":
py_type = allow_connected_value(py_type)
requires_value = self.request_requires_value
@@ -235,6 +301,7 @@ class FloatParameterModel(BaseGalaxyToolParameterModelDefinition):
value: Optional[float] = None
min: Optional[float] = None
max: Optional[float] = None
validators: List[NumberCompatiableValidators] = []
@property
def py_type(self) -> Type:
@@ -249,6 +316,10 @@ class FloatParameterModel(BaseGalaxyToolParameterModelDefinition):
requires_value = True
elif _is_landing_request(state_representation):
requires_value = False
validators = self.validators[:]
if self.min is not None or self.max is not None:
validators.append(InRangeParameterValidatorModel(min=self.min, max=self.max, implicit=True))
py_type = decorate_type_with_validators_if_needed(py_type, validators)
return dynamic_model_information_from_py_type(self, py_type, requires_value=requires_value)
@property
@@ -502,6 +573,7 @@ class DataCollectionParameterModel(BaseGalaxyToolParameterModelDefinition):
class HiddenParameterModel(BaseGalaxyToolParameterModelDefinition):
parameter_type: Literal["gx_hidden"] = "gx_hidden"
value: Optional[str]
validators: List[TextCompatiableValidators] = []
@property
def py_type(self) -> Type:
@@ -510,6 +582,7 @@ class HiddenParameterModel(BaseGalaxyToolParameterModelDefinition):
def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
py_type = self.py_type
requires_value = not self.optional and self.value is None
py_type = decorate_type_with_validators_if_needed(py_type, self.validators)
if state_representation == "workflow_step_linked":
py_type = allow_connected_value(py_type)
elif state_representation == "workflow_step" and not self.optional:
@@ -618,16 +691,21 @@ class BooleanParameterModel(BaseGalaxyToolParameterModelDefinition):
class DirectoryUriParameterModel(BaseGalaxyToolParameterModelDefinition):
parameter_type: Literal["gx_directory_uri"] = "gx_directory_uri"
validators: List[TextCompatiableValidators] = []
@property
def py_type(self) -> Type:
return AnyUrl
def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
py_type = self.py_type
py_type = decorate_type_with_validators_if_needed(py_type, self.validators)
if state_representation == "workflow_step_linked":
py_type = allow_connected_value(py_type)
requires_value = self.request_requires_value
if _is_landing_request(state_representation):
requires_value = False
return dynamic_model_information_from_py_type(self, self.py_type, requires_value=requires_value)
return dynamic_model_information_from_py_type(self, py_type, requires_value=requires_value)
@property
def request_requires_value(self) -> bool:
@@ -659,10 +737,14 @@ class RulesParameterModel(BaseGalaxyToolParameterModelDefinition):
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:
@@ -699,28 +781,30 @@ class SelectParameterModel(BaseGalaxyToolParameterModelDefinition):
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
py_type = decorate_type_with_validators_if_needed(py_type, self.validators)
return dynamic_model_information_from_py_type(
self, py_type, validators=validators, requires_value=requires_value
)
@property
def has_selected_static_option(self):
lib/galaxy/tool_util/parser/interface.py
+2 -1
View File
@@ -28,6 +28,7 @@ from typing_extensions import (
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 @@ class InputSource(metaclass=ABCMeta):
"""
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.
lib/galaxy/tool_util/parser/parameter_validators.py
+725
View File
@@ -0,0 +1,725 @@
import json
from typing import (
Any,
cast,
List,
Optional,
Sequence,
Union,
)
from pydantic import (
BaseModel,
ConfigDict,
Field,
model_validator,
PrivateAttr,
)
from typing_extensions import (
Annotated,
get_args,
Literal,
Protocol,
Self,
)
from galaxy.util import (
asbool,
Element,
)
try:
import regex
except ImportError:
import re as regex
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
# track validators setup by other input parameters and not validation explicitly
implicit: bool = False
_static: bool = PrivateAttr(False)
_deprecated: bool = PrivateAttr(False)
@model_validator(mode="after")
def set_default_message(self) -> Self:
if self.message is None:
self.message = self.default_message
return self
@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, "<string>", "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:
if value and not isinstance(value, str):
raise ValueError(f"Wrong type found value {value}")
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."
return message
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 not self.negate 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: Union[int, str]
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: Union[int, str]
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: Union[int, str]
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: Union[int, str]
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: Union[int, str]
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")
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
def static_tool_validators(validators: Sequence[ParameterValidatorModel]) -> List[StaticValidatorModel]:
static_validators: List[StaticValidatorModel] = []
for validator in validators:
if isinstance(validator, StaticValidatorModel):
static_validators.append(validator)
return static_validators
def statically_validates(validators: Sequence[ParameterValidatorModel], value: Any) -> bool:
for validator in static_tool_validators(validators):
try:
validator.statically_validate(value)
except ValueError:
return False
return True
lib/galaxy/tool_util/parser/util.py
+24
View File
@@ -8,6 +8,9 @@ from typing import (
from packaging.version import Version
from galaxy.util import string_as_bool
from .parameter_validators import statically_validates
if TYPE_CHECKING:
from .interface import (
InputSource,
@@ -83,6 +86,27 @@ def boolean_true_and_false_values(input_source, profile: Optional[Union[float, s
return (truevalue, falsevalue)
def text_input_is_optional(input_source: "InputSource") -> Tuple[bool, bool]:
# Optionality not explicitly defined, default to False
optional: Optional[bool] = False
optionality_inferred: bool = False
optional = input_source.get("optional", None)
if optional is not None:
optional = string_as_bool(optional)
else:
# A text parameter that doesn't raise a validation error on empty string
# is considered to be optional
if statically_validates(input_source.parse_validators(), ""):
optional = True
optionality_inferred = True
else:
optional = False
assert isinstance(optional, bool)
return optional, optionality_inferred
class ParameterParseException(Exception):
message: str
lib/galaxy/tool_util/parser/xml.py
+6 -2
View File
@@ -71,6 +71,10 @@ from .output_objects import (
ToolOutputCollection,
ToolOutputCollectionStructure,
)
from .parameter_validators import (
AnyValidatorModel,
parse_xml_validators,
)
from .stdio import (
aggressive_error_checks,
error_on_exit_code,
@@ -1339,8 +1343,8 @@ class XmlInputSource(InputSource):
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."""
lib/galaxy/tool_util/unittest_utils/sample_data.py
+53
View File
@@ -17,3 +17,56 @@ SIMPLE_MACRO = Template(
</macros>
"""
)
VALID_XML_VALIDATORS = [
"""<validator type="empty_dataset" />""",
"""<validator type="empty_dataset" message="foobar" />""",
"""<validator type="empty_dataset" message="foobar" negate="true" />""",
"""<validator type="expression">value == 7</validator>""",
"""<validator type="regex">mycoolexpression</validator>""",
"""<validator type="in_range" min="3.1" max="3.8" />""",
"""<validator type="in_range" min="3.1" />""",
"""<validator type="in_range" min="3.1" max="3.8" exclude_min="false" exclude_max="true" />""",
"""<validator type="length" min="3" />""",
"""<validator type="length" max="7" />""",
"""<validator type="length" min="2" max="7" negate="true" />""",
"""<validator type="empty_field" />""",
"""<validator type="empty_extra_files_path" />""",
"""<validator type="no_options" />""",
"""<validator type="unspecified_build" />""",
"""<validator type="dataset_ok_validator" />""",
"""<validator type="metadata" check="foo, bar" />""",
"""<validator type="metadata" skip="name, dbkey" />""",
"""<validator type="dataset_metadata_equal" metadata_name="foobar" value="moocow" />""",
"""<validator type="dataset_metadata_equal" metadata_name="foobar" value_json="null" />""",
"""<validator type="dataset_metadata_in_range" metadata_name="foobar" min="4.5" max="7.8" />""",
"""<validator type="dataset_metadata_in_range" metadata_name="foobar" min="4.5" max="7.8" include_min="true" />""",
"""<validator type="dataset_metadata_in_data_table" metadata_name="foobar" metadata_column="3" table_name="mycooltable" />""",
"""<validator type="dataset_metadata_not_in_data_table" metadata_name="foobar" metadata_column="3" table_name="mycooltable" />""",
"""<validator type="value_in_data_table" metadata_column="3" table_name="mycooltable" />""",
"""<validator type="value_in_data_table" table_name="mycooltable" />""",
"""<validator type="value_not_in_data_table" metadata_column="3" table_name="mycooltable" />""",
"""<validator type="value_not_in_data_table" table_name="mycooltable" />""",
]
INVALID_XML_VALIDATORS = [
"""<validator type="unknown" />""",
"""<validator type="empty_datasetx" />""",
"""<validator type="expression" />""",
"""<validator type="empty_dataset" message="foobar" negate="NOTABOOLVALUE" />""",
"""<validator type="regex" />""",
"""<validator type="in_range" min="3.1" max="3.8" exclude_min="false" exclude_max="foobar" />""",
"""<validator type="in_range" min="notanumber" max="3.8" />""",
"""<validator type="length" min="notanumber" />""",
"""<validator type="length" max="notanumber" />""",
"""<validator type="length" min="2" max="7" negate="notabool" />""",
"""<validator type="dataset_metadata_equal" />""",
"""<validator type="dataset_metadata_equal" metadaata_name="foobar" />""",
"""<validator type="dataset_metadata_equal" metadaata_name="foobar" value_json="undefined" />""",
"""<validator type="dataset_metadata_in_range" metadata_name="foobar" min="4.5" max="7.8" include_min="notabool" />"""
"""<validator type="dataset_metadata_in_range" min="4.5" max="7.8" />"""
"""<validator type="dataset_metadata_in_data_table" metadata_name="foobar" metadata_column="3" />""",
"""<validator type="dataset_metadata_in_data_table" metadata_column="3" table_name="mycooltable" />""",
"""<validator type="dataset_metadata_not_in_data_table" metadata_name="foobar" metadata_column="3" />""",
"""<validator type="dataset_metadata_not_in_data_table" metadata_column="3" table_name="mycooltable" />""",
]
lib/galaxy/tools/parameters/basic.py
+14 -21
View File
@@ -47,6 +47,7 @@ from galaxy.tool_util.parser.util import (
boolean_is_checked,
boolean_true_and_false_values,
ParameterParseException,
text_input_is_optional,
)
from galaxy.tools.parameters.workflow_utils import workflow_building_modes
from galaxy.util import (
@@ -194,9 +195,7 @@ class ToolParameter(UsesDictVisibleKeys):
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 if tool else None, input_source.parse_validators())
@property
def visible(self) -> bool:
@@ -351,7 +350,6 @@ class ToolParameter(UsesDictVisibleKeys):
class SimpleTextToolParameter(ToolParameter):
def __init__(self, tool, input_source):
input_source = ensure_input_source(input_source)
self.optionality_inferred = False
super().__init__(tool, input_source)
optional = input_source.get("optional", None)
if optional is not None:
@@ -359,18 +357,7 @@ class SimpleTextToolParameter(ToolParameter):
else:
# Optionality not explicitly defined, default to False
optional = False
if self.type == "text":
# A text parameter that doesn't raise a validation error on empty string
# is considered to be optional
try:
for validator in self.validators:
validator.validate("")
optional = True
self.optionality_inferred = True
except ValueError:
pass
self.optional = optional
if self.optional:
self.value = None
else:
@@ -405,10 +392,16 @@ class TextToolParameter(SimpleTextToolParameter):
def __init__(self, tool, input_source):
input_source = ensure_input_source(input_source)
super().__init__(tool, input_source)
self.profile = tool.profile
self.profile = tool.profile if tool else None
self.datalist = []
for title, value, _ in input_source.parse_static_options():
self.datalist.append({"label": title, "value": value})
# why does Integer and Float subclass this :_(
if self.type == "text":
self.optional, self.optionality_inferred = text_input_is_optional(input_source)
else:
self.optionality_inferred = False
self.value = input_source.get("value")
self.area = input_source.get_bool("area", False)
@@ -422,10 +415,10 @@ class TextToolParameter(SimpleTextToolParameter):
return super().validate(value, trans)
@property
def wrapper_default() -> Optional[str]:
def wrapper_default(self) -> Optional[str]:
"""Handle change in default handling pre and post 23.0 profiles."""
profile = self.profile
legacy_behavior = (profile is None or Version(str(profile)) < Version("23.0"))
legacy_behavior = profile is None or Version(str(profile)) < Version("23.0")
default_value = None
if self.optional and self.optionality_inferred and legacy_behavior:
default_value = ""
@@ -478,7 +471,7 @@ class IntegerToolParameter(TextToolParameter):
except ValueError:
raise ParameterValueError("attribute 'max' must be an integer", self.name, self.max)
if self.min is not None or self.max is not None:
self.validators.append(validation.InRangeValidator(None, self.min, self.max))
self.validators.append(validation.InRangeValidator.simple_range_validator(self.min, self.max))
def from_json(self, value, trans, other_values=None):
other_values = other_values or {}
@@ -551,7 +544,7 @@ class FloatToolParameter(TextToolParameter):
except ValueError:
raise ParameterValueError("attribute 'max' must be a real number", self.name, self.max)
if self.min is not None or self.max is not None:
self.validators.append(validation.InRangeValidator(None, self.min, self.max))
self.validators.append(validation.InRangeValidator.simple_range_validator(self.min, self.max))
def from_json(self, value, trans, other_values=None):
other_values = other_values or {}
@@ -2056,7 +2049,7 @@ class DataToolParameter(BaseDataToolParameter):
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)
lib/galaxy/tools/parameters/dynamic_options.py
+3 -2
View File
@@ -621,8 +621,9 @@ class DynamicOptions:
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
lib/galaxy/tools/parameters/validation.py
+147 -251
View File
@@ -3,16 +3,30 @@ Classes related to parameter validation.
"""
import abc
import json
import logging
import os.path
import regex
import os
from typing import (
Any,
cast,
List,
Optional,
Union,
)
from galaxy import (
model,
util,
)
from galaxy.tool_util.parser.parameter_validators import (
AnyValidatorModel,
EmptyFieldParameterValidatorModel,
ExpressionParameterValidatorModel,
InRangeParameterValidatorModel,
MetadataParameterValidatorModel,
parse_xml_validators as parse_xml_validators_models,
raise_error_if_valiation_fails,
RegexParameterValidatorModel,
)
log = logging.getLogger(__name__)
@@ -24,27 +38,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 +62,7 @@ class Validator(abc.ABC):
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 +70,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 +85,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, "<string>", "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 +102,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,
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 +118,10 @@ class InRangeValidator(ExpressionValidator):
(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,24 +131,23 @@ class InRangeValidator(ExpressionValidator):
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)
@staticmethod
def simple_range_validator(min: Optional[float], max: Optional[float]):
return cast(
InRangeParameterValidatorModel,
_to_validator(None, InRangeParameterValidatorModel(min=min, max=max, implicit=True)),
)
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 +160,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 +173,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 +186,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 +201,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 cast(MetadataValidator, _to_validator(None, MetadataParameterValidatorModel(implicit=True)))
def validate(self, value, trans=None):
if value:
@@ -293,25 +234,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 +251,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 +269,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 +281,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,35 +299,19 @@ 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,
line_startswith: Optional[str] = None,
split: str = "\t",
negate: bool = False,
):
super().__init__(message, negate)
assert filename
assert os.path.exists(filename), f"File {filename} specified by the 'filename' attribute not found"
self.metadata_name = metadata_name
self.valid_values = set()
with open(filename) as fh:
@@ -445,28 +337,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 +380,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 +403,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 +431,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 +458,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,
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 +508,29 @@ validator_types = dict(
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]) -> List[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")
del as_dict["implicit"]
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
if "filename" in as_dict and app is not None:
filename = as_dict.pop("filename")
as_dict["filename"] = f"{app.config.tool_data_path}/{filename}"
return validator_types[validator_type](**as_dict)
lib/galaxy/tools/wrappers.py
+1 -1
View File
@@ -130,7 +130,7 @@ class InputValueWrapper(ToolParameterValueWrapper):
self.input = input
if value is None and input.type == "text":
# Tools with old profile versions may treat an optional text parameter as `""`
value = cast(TextToolParameter, input).wrapper_default()
value = cast(TextToolParameter, input).wrapper_default
self.value = value
self._other_values: Dict[str, str] = other_values or {}
test/functional/tools/parameters/gx_directory_uri_validation.xml
+16
View File
@@ -0,0 +1,16 @@
<tool id="gx_directory_uri" name="gx_directory_uri" version="1.0.0">
<macros>
<import>macros.xml</import>
</macros>
<command><![CDATA[
echo 'parameter: $parameter' > $output
]]></command>
<inputs>
<param type="directory_uri" name="parameter">
<validator type="length" min="10" max="45" />
<validator type="expression">'api' in value</validator>
<validator type="regex">^.*json$</validator>
</param>
</inputs>
<expand macro="simple_text_output" />
</tool>
test/functional/tools/parameters/gx_float_min_max.xml
+14
View File
@@ -0,0 +1,14 @@
<tool id="gx_float_validation_range" name="gx_float_validation_range" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output'
]]></command>
<inputs>
<param name="parameter" value="1" type="float" min="0.1" max="9.8">
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
</outputs>
<tests>
</tests>
</tool>
test/functional/tools/parameters/gx_float_validation_range.xml
+15
View File
@@ -0,0 +1,15 @@
<tool id="gx_float_validation_range" name="gx_float_validation_range" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output'
]]></command>
<inputs>
<param name="parameter" value="1" type="float">
<validator type="in_range" min="0.1" max="9.8" exclude_min="false" exclude_max="true" />
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
</outputs>
<tests>
</tests>
</tool>
test/functional/tools/parameters/gx_hidden_validation.xml
+17
View File
@@ -0,0 +1,17 @@
<tool id="gx_hidden_validation" name="gx_hidden_validation" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output'
]]></command>
<inputs>
<param name="parameter" type="hidden" value="http://mycoolsource/api/index.json">
<validator type="length" min="10" max="45" />
<validator type="expression">'api' in value</validator>
<validator type="regex">^.*json$</validator>
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
</outputs>
<tests>
</tests>
</tool>
test/functional/tools/parameters/gx_int_min_max.xml
+14
View File
@@ -0,0 +1,14 @@
<tool id="gx_int_min_max" name="gx_int_min_max" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output'
]]></command>
<inputs>
<param name="parameter" value="1" type="integer" min="0" max="10">
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
</outputs>
<tests>
</tests>
</tool>
test/functional/tools/parameters/gx_int_validation_range.xml
+15
View File
@@ -0,0 +1,15 @@
<tool id="gx_int_validation_range" name="gx_int_validation_range" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output'
]]></command>
<inputs>
<param name="parameter" value="1" type="integer">
<validator type="in_range" min="0" max="10" exclude_min="false" exclude_max="true" />
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
</outputs>
<tests>
</tests>
</tool>
test/functional/tools/parameters/gx_text_expression_validation.xml
+20
View File
@@ -0,0 +1,20 @@
<tool id="gx_text_expression_validation" name="gx_text_expression_validation" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output';
cat '$inputs' >> $inputs_json;
]]></command>
<configfiles>
<inputs name="inputs" filename="inputs.json" />
</configfiles>
<inputs>
<param name="parameter" type="text">
<validator type="expression">'foobar' in value</validator>
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
<data name="inputs_json" format="json" />
</outputs>
<tests>
</tests>
</tool>
test/functional/tools/parameters/gx_text_length_validation.xml
+20
View File
@@ -0,0 +1,20 @@
<tool id="gx_text_length_validation" name="gx_text_length_validation" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output';
cat '$inputs' >> $inputs_json;
]]></command>
<configfiles>
<inputs name="inputs" filename="inputs.json" />
</configfiles>
<inputs>
<param name="parameter" type="text">
<validator type="length" min="2" max="10" />
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
<data name="inputs_json" format="json" />
</outputs>
<tests>
</tests>
</tool>
test/functional/tools/parameters/gx_text_length_validation_negate.xml
+20
View File
@@ -0,0 +1,20 @@
<tool id="gx_text_length_validation" name="gx_text_length_validation" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output';
cat '$inputs' >> $inputs_json;
]]></command>
<configfiles>
<inputs name="inputs" filename="inputs.json" />
</configfiles>
<inputs>
<param name="parameter" type="text">
<validator type="length" min="2" max="10" negate="true" />
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
<data name="inputs_json" format="json" />
</outputs>
<tests>
</tests>
</tool>
test/functional/tools/parameters/gx_text_regex_validation.xml
+20
View File
@@ -0,0 +1,20 @@
<tool id="gx_text_regex_validation" name="gx_text_regex_validation" version="1.0.0">
<command><![CDATA[
echo '$parameter' >> '$output';
cat '$inputs' >> $inputs_json;
]]></command>
<configfiles>
<inputs name="inputs" filename="inputs.json" />
</configfiles>
<inputs>
<param name="parameter" type="text">
<validator type="regex">^[actg]*$</validator>
</param>
</inputs>
<outputs>
<data name="output" format="txt" />
<data name="inputs_json" format="json" />
</outputs>
<tests>
</tests>
</tool>
test/unit/app/tools/test_dynamic_options.py
+5 -1
View File
@@ -6,6 +6,10 @@ from galaxy.work.context import WorkRequestContext
def get_from_url_option():
tool_param = Bunch()
tool_param.tool = Bunch()
tool_param.tool.app = Bunch()
return DynamicOptions(
XML(
"""
@@ -26,7 +30,7 @@ def get_from_url_option():
</options>
"""
),
Bunch(),
tool_param,
)
test/unit/app/tools/test_parameter_validation.py
+2 -2
View File
@@ -220,12 +220,12 @@ class TestParameterValidation(BaseParameterTestCase):
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)
test/unit/app/tools/test_validation_parsing.py
+41
View File
@@ -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"<param>{xml_str}</param>")
parse_xml_validators(MockApp(), xml_el)
test/unit/tool_util/parameter_specification.yml
+143 -4
View File
@@ -191,15 +191,81 @@ 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: {}
- parameter: {__class__: 'ConnectedValue'}
- parameter: -1
- parameter: 11
- parameter: 10
workflow_step_linked_valid:
- parameter: {__class__: 'ConnectedValue'}
gx_int_min_max:
request_valid:
- parameter: 1
- parameter: 9
- parameter: 10
request_invalid:
- parameter: -1
- parameter: 11
workflow_step_linked_valid:
- parameter: {__class__: 'ConnectedValue'}
workflow_step_linked_invalid:
- parameter: -1
gx_text_regex_validation:
request_valid:
- parameter: acgt
- parameter: a
- parameter: aaaggttac
request_invalid:
- parameter: acgu
- parameter: nucleic
workflow_step_linked_valid:
- parameter: {__class__: 'ConnectedValue'}
gx_text_expression_validation:
request_valid:
- parameter: the code was foobar
- parameter: foobar
request_invalid:
- parameter: the code was not foo bar
- parameter: ''
workflow_step_linked_valid:
- parameter: {__class__: 'ConnectedValue'}
workflow_step_linked_invalid:
- parameter: ''
gx_text_empty_validation:
request_valid:
- parameter: foobar
- {}
request_invalid:
- parameter: ''
job_internal_valid:
- parameter: foobar
job_internal_invalid:
- {}
- parameter: ''
workflow_step_linked_valid:
- parameter: {__class__: 'ConnectedValue'}
workflow_step_linked_invalid:
- parameter: ''
gx_text:
request_valid: &gx_text_request_valid
- parameter: moocow
- parameter: 'some spaces'
- parameter: ''
- {}
# need to explicitly mark these as non-optional
- parameter: null
request_invalid: &gx_text_request_invalid
- parameter: 5
- parameter: null
- parameter: {}
- parameter: { "moo": "cow" }
- parameter: {__class__: 'ConnectedValue'}
@@ -223,18 +289,18 @@ gx_text:
- parameter: moocow
- parameter: 'some spaces'
- parameter: ''
- parameter: null
job_internal_invalid:
- {}
- parameter: null
- parameter: { "moo": "cow" }
workflow_step_valid:
- parameter: moocow
- parameter: 'some spaces'
- parameter: ''
- {}
- parameter: null
workflow_step_invalid:
- parameter: 5
- parameter: null
- parameter: {}
- parameter: { "moo": "cow" }
- parameter: {__class__: 'ConnectedValue'}
@@ -244,9 +310,9 @@ gx_text:
- parameter: ''
- {}
- parameter: {__class__: 'ConnectedValue'}
- parameter: null
workflow_step_linked_invalid:
- parameter: 5
- parameter: null
- parameter: {}
- parameter: { "moo": "cow" }
- parameter: {"class": 'ConnectedValue'}
@@ -294,6 +360,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 +565,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
@@ -571,6 +658,22 @@ gx_directory_uri:
- parameter: true
- parameter: null
gx_directory_uri_validation:
request_valid:
- parameter: "gxfiles://mycoolsource/api/index.json"
request_invalid:
- parameter: {__class__: 'ConnectedValue'}
- parameter: "gxfiles://tooshort/index.json"
- parameter: "gxfiles://mycoolsource/api/wrongex.js"
- parameter: "gxfiles://mycoolsource/badexp.json"
workflow_step_linked_valid:
- parameter: {__class__: 'ConnectedValue'}
- parameter: "gxfiles://mycoolsource/api/index.json"
workflow_step_linked_invalid:
- parameter: "gxfiles://tooshort/index.json"
- parameter: "gxfiles://mycoolsource/api/wrongex.js"
- parameter: "gxfiles://mycoolsource/badexp.json"
gx_hidden:
request_valid:
- parameter: moocow
@@ -656,6 +759,22 @@ gx_hidden_optional:
workflow_step_linked_invalid:
- parameter: 5
gx_hidden_validation:
request_valid:
- parameter: "http://mycoolservice.com/api/index.json"
request_invalid:
- parameter: {__class__: 'ConnectedValue'}
- parameter: "http://mycoolservice.com/index.json"
- parameter: "http://mycoolservice.com/api/wrongex.js"
- parameter: "http://mycoolservice.com/badexp.json"
workflow_step_linked_valid:
- parameter: {__class__: 'ConnectedValue'}
- parameter: "http://mycoolservice.com/api/index.json"
workflow_step_linked_invalid:
- parameter: "http://mycoolservice.com/index.json"
- parameter: "http://mycoolservice.com/api/wrongex.js"
- parameter: "http://mycoolservice.com/badexp.json"
gx_float:
request_valid:
- parameter: 5
@@ -779,6 +898,26 @@ gx_float_optional:
- parameter: 'foobar'
- parameter: {__class__: 'ConnectedValue2'}
gx_float_validation_range:
request_valid:
- parameter: 0.1
- parameter: 9.7
- parameter: 5
request_invalid:
- parameter: 10
- parameter: 0
- parameter: 9.8 # endpoint not included because specified in validator
gx_float_min_max:
request_valid:
- parameter: 0.1
- parameter: 9.7
- parameter: 5
- parameter: 9.8 # endpoint implicitly included when max attribute set on param
request_invalid:
- parameter: 10
- parameter: 0
gx_color:
request_valid:
- parameter: '#aabbcc'
test/unit/tool_util/test_parameter_validator_models.py
+28
View File
@@ -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"<param>{xml_str}</param>")
parse_xml_validators(xml_el)