diff --git a/Makefile b/Makefile index ec4b04e0860c..fbfd14558e1d 100644 --- a/Makefile +++ b/Makefile @@ -186,6 +186,9 @@ else $(IN_VENV) cd client && yarn install $(YARN_INSTALL_OPTS) endif +format-xsd: + xmllint --format --output galaxy-tmp.xsd lib/galaxy/tool_util/xsd/galaxy.xsd + mv galaxy-tmp.xsd lib/galaxy/tool_util/xsd/galaxy.xsd build-api-schema: $(IN_VENV) python scripts/dump_openapi_schema.py _schema.yaml diff --git a/doc/parse_gx_xsd.py b/doc/parse_gx_xsd.py index e34a9ba89afd..11092bb84515 100644 --- a/doc/parse_gx_xsd.py +++ b/doc/parse_gx_xsd.py @@ -83,28 +83,20 @@ def _build_tag(tag, hide_attributes): assertions_buffer.write(_doc_or_none(assertions_tag)) assertions_buffer.write("\n\n") - assertion_groups = assertions_tag.xpath( - "xs:choice/xs:group", namespaces={"xs": "http://www.w3.org/2001/XMLSchema"} + assertion_tag = xmlschema_doc.find("//{http://www.w3.org/2001/XMLSchema}group[@name='TestAssertion']") + elements = assertion_tag.findall( + "{http://www.w3.org/2001/XMLSchema}choice/{http://www.w3.org/2001/XMLSchema}element" ) - for group in assertion_groups: - ref = group.attrib["ref"] - assertion_tag = xmlschema_doc.find("//{http://www.w3.org/2001/XMLSchema}group[@name='" + ref + "']") - doc = _doc_or_none(assertion_tag) - assertions_buffer.write(f"### {doc}\n\n") - elements = assertion_tag.findall( - "{http://www.w3.org/2001/XMLSchema}choice/{http://www.w3.org/2001/XMLSchema}element" - ) - for element in elements: + for element in elements: + doc = _doc_or_none(element) + if doc is None: doc = _doc_or_none(element) - if doc is None: - doc = _doc_or_none(_type_el(element)) - assert doc is not None, f"Documentation for {element.attrib['name']} is empty" - doc = doc.strip() - - element_el = _find_tag_el(element) - element_attributes = _find_attributes(element_el) - doc = _replace_attribute_list(element_el, doc, element_attributes) - assertions_buffer.write(f"#### ``{element.attrib['name']}``:\n\n{doc}\n\n") + assert doc is not None, f"Documentation for {element.attrib['name']} is empty" + doc = doc.strip() + + element_attributes = _find_attributes(element) + doc = _replace_attribute_list(element, doc, element_attributes) + assertions_buffer.write(f"#### ``{element.attrib['name']}``:\n\n{doc}\n\n") text = text.replace(line, assertions_buffer.getvalue()) tag_help.write(text) if best_practices := _get_bp_link(annotation_el): diff --git a/lib/galaxy/tool_util/verify/assertion_models.py b/lib/galaxy/tool_util/verify/assertion_models.py new file mode 100644 index 000000000000..5f21e488e52b --- /dev/null +++ b/lib/galaxy/tool_util/verify/assertion_models.py @@ -0,0 +1,1599 @@ +"""This module is auto-generated, please do not modify.""" + +import re +import typing + +from pydantic import ( + BaseModel, + BeforeValidator, + ConfigDict, + Field, + RootModel, + StrictFloat, + StrictInt, +) +from typing_extensions import ( + Annotated, + Literal, +) + +BYTES_PATTERN = re.compile(r"^(0|[1-9][0-9]*)([kKMGTPE]i?)?$") + + +class AssertionModel(BaseModel): + model_config = ConfigDict( + extra="forbid", + ) + + +def check_bytes(v: typing.Any) -> typing.Any: + if isinstance(v, str): + assert BYTES_PATTERN.match(v), "Not valid bytes string" + return v + + +def check_center_of_mass(v: typing.Any): + assert isinstance(v, str) + split_parts = v.split(",") + assert len(split_parts) == 2 + for part in split_parts: + assert float(part.strip()) + return v + + +def check_regex(v: typing.Any): + assert isinstance(v, str) + try: + re.compile(typing.cast(str, v)) + except re.error: + raise AssertionError(f"Invalid regular expression {v}") + return v + + +def check_non_negative_if_set(v: typing.Any): + if v is not None: + try: + assert v >= 0 + except TypeError: + raise AssertionError(f"Invalid type found {v}") + return v + + +def check_non_negative_if_int(v: typing.Any): + if v is not None and isinstance(v, int): + assert typing.cast(int, v) >= 0 + return v + + +has_line_line_description = """The full line of text to search for in the output.""" + +has_line_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_line_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_line_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_line_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_line_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_line_model(AssertionModel): + r"""Asserts the specified output contains the line specified by the + argument line. The exact number of occurrences can be optionally + specified by the argument n""" + + that: Literal["has_line"] = "has_line" + + line: str = Field( + ..., + description=has_line_line_description, + ) + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_line_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_line_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_line_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_line_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_line_negate_description, + ) + + +has_line_matching_expression_description = """The regular expressions to attempt match in the output.""" + +has_line_matching_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_line_matching_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_line_matching_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_line_matching_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_line_matching_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_line_matching_model(AssertionModel): + r"""Asserts the specified output contains a line matching the + regular expression specified by the argument expression. If n is given + the assertion checks for exactly n occurences.""" + + that: Literal["has_line_matching"] = "has_line_matching" + + expression: str = Field( + ..., + description=has_line_matching_expression_description, + ) + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_line_matching_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_line_matching_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_line_matching_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_line_matching_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_line_matching_negate_description, + ) + + +has_n_lines_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_n_lines_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_n_lines_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_n_lines_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_n_lines_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_n_lines_model(AssertionModel): + r"""Asserts the specified output contains ``n`` lines allowing + for a difference in the number of lines (delta) + or relative differebce in the number of lines""" + + that: Literal["has_n_lines"] = "has_n_lines" + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_lines_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_n_lines_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_lines_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_lines_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_n_lines_negate_description, + ) + + +has_text_text_description = """The text to search for in the output.""" + +has_text_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_text_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_text_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_text_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_text_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_text_model(AssertionModel): + r"""Asserts specified output contains the substring specified by + the argument text. The exact number of occurrences can be + optionally specified by the argument n""" + + that: Literal["has_text"] = "has_text" + + text: str = Field( + ..., + description=has_text_text_description, + ) + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_text_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_text_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_text_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_text_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_text_negate_description, + ) + + +has_text_matching_expression_description = """The regular expressions to attempt match in the output.""" + +has_text_matching_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_text_matching_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_text_matching_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_text_matching_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_text_matching_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_text_matching_model(AssertionModel): + r"""Asserts the specified output contains text matching the + regular expression specified by the argument expression. + If n is given the assertion checks for exacly n (nonoverlapping) + occurences.""" + + that: Literal["has_text_matching"] = "has_text_matching" + + expression: str = Field( + ..., + description=has_text_matching_expression_description, + ) + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_text_matching_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_text_matching_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_text_matching_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_text_matching_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_text_matching_negate_description, + ) + + +not_has_text_text_description = """The text to search for in the output.""" + + +class not_has_text_model(AssertionModel): + r"""Asserts specified output does not contain the substring + specified by the argument text""" + + that: Literal["not_has_text"] = "not_has_text" + + text: str = Field( + ..., + description=not_has_text_text_description, + ) + + +has_n_columns_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_n_columns_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_n_columns_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_n_columns_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_n_columns_sep_description = """Separator defining columns, default: tab""" + +has_n_columns_comment_description = ( + """Comment character(s) used to skip comment lines (which should not be used for counting columns)""" +) + +has_n_columns_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_n_columns_model(AssertionModel): + r"""Asserts tabular output contains the specified + number (``n``) of columns. + + For instance, ````. The assertion tests only the first line. + Number of columns can optionally also be specified with ``delta``. Alternatively the + range of expected occurences can be specified by ``min`` and/or ``max``. + + Optionally a column separator (``sep``, default is `` ``) `and comment character(s) + can be specified (``comment``, default is empty string). The first non-comment + line is used for determining the number of columns.""" + + that: Literal["has_n_columns"] = "has_n_columns" + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_columns_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_n_columns_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_columns_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_columns_max_description, + ) + + sep: str = Field( + " ", + description=has_n_columns_sep_description, + ) + + comment: str = Field( + "", + description=has_n_columns_comment_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_n_columns_negate_description, + ) + + +attribute_is_path_description = """The Python xpath-like expression to find the target element.""" + +attribute_is_attribute_description = """The XML attribute name to test against from the target XML element.""" + +attribute_is_text_description = """The expected attribute value to test against on the target XML element""" + +attribute_is_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class attribute_is_model(AssertionModel): + r"""Asserts the XML ``attribute`` for the element (or tag) with the specified + XPath-like ``path`` is the specified ``text``. + + For example: + + ```xml + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the assertion (on the equality) can be inverted (the + implicit assertion on the existence of the path is not affected).""" + + that: Literal["attribute_is"] = "attribute_is" + + path: str = Field( + ..., + description=attribute_is_path_description, + ) + + attribute: str = Field( + ..., + description=attribute_is_attribute_description, + ) + + text: str = Field( + ..., + description=attribute_is_text_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=attribute_is_negate_description, + ) + + +attribute_matches_path_description = """The Python xpath-like expression to find the target element.""" + +attribute_matches_attribute_description = """The XML attribute name to test against from the target XML element.""" + +attribute_matches_expression_description = ( + """The regular expressions to apply against the named attribute on the target XML element.""" +) + +attribute_matches_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class attribute_matches_model(AssertionModel): + r"""Asserts the XML ``attribute`` for the element (or tag) with the specified + XPath-like ``path`` matches the regular expression specified by ``expression``. + + For example: + + ```xml + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the assertion (on the matching) can be inverted (the + implicit assertion on the existence of the path is not affected).""" + + that: Literal["attribute_matches"] = "attribute_matches" + + path: str = Field( + ..., + description=attribute_matches_path_description, + ) + + attribute: str = Field( + ..., + description=attribute_matches_attribute_description, + ) + + expression: Annotated[str, BeforeValidator(check_regex)] = Field( + ..., + description=attribute_matches_expression_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=attribute_matches_negate_description, + ) + + +element_text_path_description = """The Python xpath-like expression to find the target element.""" + +element_text_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class element_text_model(AssertionModel): + r"""This tag allows the developer to recurisively specify additional assertions as + child elements about just the text contained in the element specified by the + XPath-like ``path``, e.g. + + ```xml + + + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the implicit assertions can be inverted. + The sub-assertions, which have their own ``negate`` attribute, are not affected + by ``negate``.""" + + that: Literal["element_text"] = "element_text" + + path: str = Field( + ..., + description=element_text_path_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=element_text_negate_description, + ) + + children: "assertion_list" + + +element_text_is_path_description = """The Python xpath-like expression to find the target element.""" + +element_text_is_text_description = ( + """The expected element text (body of the XML tag) to test against on the target XML element""" +) + +element_text_is_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class element_text_is_model(AssertionModel): + r"""Asserts the text of the XML element with the specified XPath-like ``path`` is + the specified ``text``. + + For example: + + ```xml + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the assertion (on the equality) can be inverted (the + implicit assertion on the existence of the path is not affected).""" + + that: Literal["element_text_is"] = "element_text_is" + + path: str = Field( + ..., + description=element_text_is_path_description, + ) + + text: str = Field( + ..., + description=element_text_is_text_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=element_text_is_negate_description, + ) + + +element_text_matches_path_description = """The Python xpath-like expression to find the target element.""" + +element_text_matches_expression_description = """The regular expressions to apply against the target element.""" + +element_text_matches_negate_description = ( + """A boolean that can be set to true to negate the outcome of the assertion.""" +) + + +class element_text_matches_model(AssertionModel): + r"""Asserts the text of the XML element with the specified XPath-like ``path`` + matches the regular expression defined by ``expression``. + + For example: + + ```xml + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the assertion (on the matching) can be inverted (the + implicit assertion on the existence of the path is not affected).""" + + that: Literal["element_text_matches"] = "element_text_matches" + + path: str = Field( + ..., + description=element_text_matches_path_description, + ) + + expression: Annotated[str, BeforeValidator(check_regex)] = Field( + ..., + description=element_text_matches_expression_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=element_text_matches_negate_description, + ) + + +has_element_with_path_path_description = """The Python xpath-like expression to find the target element.""" + +has_element_with_path_negate_description = ( + """A boolean that can be set to true to negate the outcome of the assertion.""" +) + + +class has_element_with_path_model(AssertionModel): + r"""Asserts the XML output contains at least one element (or tag) with the specified + XPath-like ``path``, e.g. + + ```xml + + ``` + + With ``negate`` the result of the assertion can be inverted.""" + + that: Literal["has_element_with_path"] = "has_element_with_path" + + path: str = Field( + ..., + description=has_element_with_path_path_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_element_with_path_negate_description, + ) + + +has_n_elements_with_path_path_description = """The Python xpath-like expression to find the target element.""" + +has_n_elements_with_path_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_n_elements_with_path_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_n_elements_with_path_min_description = ( + """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_n_elements_with_path_max_description = ( + """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_n_elements_with_path_negate_description = ( + """A boolean that can be set to true to negate the outcome of the assertion.""" +) + + +class has_n_elements_with_path_model(AssertionModel): + r"""Asserts the XML output contains the specified number (``n``, optionally with ``delta``) of elements (or + tags) with the specified XPath-like ``path``. + + For example: + + ```xml + + ``` + + Alternatively to ``n`` and ``delta`` also the ``min`` and ``max`` attributes + can be used to specify the range of the expected number of occurences. + With ``negate`` the result of the assertion can be inverted.""" + + that: Literal["has_n_elements_with_path"] = "has_n_elements_with_path" + + path: str = Field( + ..., + description=has_n_elements_with_path_path_description, + ) + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_elements_with_path_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_n_elements_with_path_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_elements_with_path_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_n_elements_with_path_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_n_elements_with_path_negate_description, + ) + + +class is_valid_xml_model(AssertionModel): + r"""Asserts the output is a valid XML file (e.g. ````).""" + + that: Literal["is_valid_xml"] = "is_valid_xml" + + +xml_element_path_description = """The Python xpath-like expression to find the target element.""" + +xml_element_attribute_description = """The XML attribute name to test against from the target XML element.""" + +xml_element_all_description = ( + """Check the sub-assertions for all paths matching the path. Default: false, i.e. only the first """ +) + +xml_element_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +xml_element_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +xml_element_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +xml_element_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +xml_element_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class xml_element_model(AssertionModel): + r"""Assert if the XML file contains element(s) or tag(s) with the specified + [XPath-like ``path``](https://lxml.de/xpathxslt.html). If ``n`` and ``delta`` + or ``min`` and ``max`` are given also the number of occurences is checked. + + ```xml + + + + + + ``` + + With ``negate="true"`` the outcome of the assertions wrt the precence and number + of ``path`` can be negated. If there are any sub assertions then check them against + + - the content of the attribute ``attribute`` + - the element's text if no attribute is given + + ```xml + + + + + + ``` + + Sub-assertions are not subject to the ``negate`` attribute of ``xml_element``. + If ``all`` is ``true`` then the sub assertions are checked for all occurences. + + Note that all other XML assertions can be expressed by this assertion (Galaxy + also implements the other assertions by calling this one).""" + + that: Literal["xml_element"] = "xml_element" + + path: str = Field( + ..., + description=xml_element_path_description, + ) + + attribute: typing.Optional[typing.Union[str]] = Field( + None, + description=xml_element_attribute_description, + ) + + all: typing.Union[bool, str] = Field( + False, + description=xml_element_all_description, + ) + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=xml_element_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=xml_element_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=xml_element_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=xml_element_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=xml_element_negate_description, + ) + + children: typing.Optional["assertion_list"] = None + + +has_json_property_with_text_property_description = """The property name to search the JSON document for.""" + +has_json_property_with_text_text_description = """The expected text value of the target JSON attribute.""" + + +class has_json_property_with_text_model(AssertionModel): + r"""Asserts the JSON document contains a property or key with the specified text (i.e. string) value. + + ```xml + + ```""" + + that: Literal["has_json_property_with_text"] = "has_json_property_with_text" + + property: str = Field( + ..., + description=has_json_property_with_text_property_description, + ) + + text: str = Field( + ..., + description=has_json_property_with_text_text_description, + ) + + +has_json_property_with_value_property_description = """The property name to search the JSON document for.""" + +has_json_property_with_value_value_description = ( + """The expected JSON value of the target JSON attribute (as a JSON encoded string).""" +) + + +class has_json_property_with_value_model(AssertionModel): + r"""Asserts the JSON document contains a property or key with the specified JSON value. + + ```xml + + ```""" + + that: Literal["has_json_property_with_value"] = "has_json_property_with_value" + + property: str = Field( + ..., + description=has_json_property_with_value_property_description, + ) + + value: str = Field( + ..., + description=has_json_property_with_value_value_description, + ) + + +has_h5_attribute_key_description = """HDF5 attribute to check value of.""" + +has_h5_attribute_value_description = """Expected value of HDF5 attribute to check.""" + + +class has_h5_attribute_model(AssertionModel): + r"""Asserts HDF5 output contains the specified ``value`` for an attribute (``key``), e.g. + + ```xml + + ```""" + + that: Literal["has_h5_attribute"] = "has_h5_attribute" + + key: str = Field( + ..., + description=has_h5_attribute_key_description, + ) + + value: str = Field( + ..., + description=has_h5_attribute_value_description, + ) + + +has_h5_keys_keys_description = """HDF5 attributes to check value of as a comma-separated string.""" + + +class has_h5_keys_model(AssertionModel): + r"""Asserts the specified HDF5 output has the given keys.""" + + that: Literal["has_h5_keys"] = "has_h5_keys" + + keys: str = Field( + ..., + description=has_h5_keys_keys_description, + ) + + +has_archive_member_path_description = """The regular expression specifying the archive member.""" + +has_archive_member_all_description = ( + """Check the sub-assertions for all paths matching the path. Default: false, i.e. only the first""" +) + +has_archive_member_n_description = """Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_archive_member_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_archive_member_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_archive_member_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_archive_member_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_archive_member_model(AssertionModel): + r"""This tag allows to check if ``path`` is contained in a compressed file. + + The path is a regular expression that is matched against the full paths of the objects in + the compressed file (remember that "matching" means it is checked if a prefix of + the full path of an archive member is described by the regular expression). + Valid archive formats include ``.zip``, ``.tar``, and ``.tar.gz``. Note that + depending on the archive creation method: + + - full paths of the members may be prefixed with ``./`` + - directories may be treated as empty files + + ```xml + + ``` + + With ``n`` and ``delta`` (or ``min`` and ``max``) assertions on the number of + archive members matching ``path`` can be expressed. The following could be used, + e.g., to assert an archive containing n±1 elements out of which at least + 4 need to have a ``txt`` extension. + + ```xml + + + ``` + + In addition the tag can contain additional assertions as child elements about + the first member in the archive matching the regular expression ``path``. For + instance + + ```xml + + + + ``` + + If the ``all`` attribute is set to ``true`` then all archive members are subject + to the assertions. Note that, archive members matching the ``path`` are sorted + alphabetically. + + The ``negate`` attribute of the ``has_archive_member`` assertion only affects + the asserts on the presence and number of matching archive members, but not any + sub-assertions (which can offer the ``negate`` attribute on their own). The + check if the file is an archive at all, which is also done by the function, is + not affected.""" + + that: Literal["has_archive_member"] = "has_archive_member" + + path: str = Field( + ..., + description=has_archive_member_path_description, + ) + + all: typing.Union[bool, str] = Field( + False, + description=has_archive_member_all_description, + ) + + n: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_archive_member_n_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_archive_member_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_archive_member_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_archive_member_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_archive_member_negate_description, + ) + + children: typing.Optional["assertion_list"] = None + + +has_size_value_description = """Deprecated alias for `size`""" + +has_size_size_description = """Desired size of the output (in bytes), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_size_delta_description = ( + """Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``""" +) + +has_size_min_description = """Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_size_max_description = """Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``""" + +has_size_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_size_model(AssertionModel): + r"""Asserts the specified output has a size of the specified value + + Attributes size and value or synonyms though value is considered deprecated. + The size optionally allows for absolute (``delta``) difference.""" + + that: Literal["has_size"] = "has_size" + + size: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_size_size_description, + ) + + delta: Annotated[ + typing.Union[int, str], BeforeValidator(check_bytes), BeforeValidator(check_non_negative_if_int) + ] = Field( + 0, + description=has_size_delta_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_size_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[str, int]], + BeforeValidator(check_bytes), + BeforeValidator(check_non_negative_if_int), + ] = Field( + None, + description=has_size_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_size_negate_description, + ) + + +has_image_center_of_mass_center_of_mass_description = """The required center of mass of the image intensities (horizontal and vertical coordinate, separated by a comma).""" + +has_image_center_of_mass_channel_description = """Restricts the assertion to a specific channel of the image (where ``0`` corresponds to the first image channel).""" + +has_image_center_of_mass_eps_description = ( + """The maximum allowed Euclidean distance to the required center of mass (defaults to ``0.01``).""" +) + + +class has_image_center_of_mass_model(AssertionModel): + r"""Asserts the specified output is an image and has the specified center of mass. + + Asserts the output is an image and has a specific center of mass, + or has an Euclidean distance of ``eps`` or less to that point (e.g., + ````).""" + + that: Literal["has_image_center_of_mass"] = "has_image_center_of_mass" + + center_of_mass: Annotated[str, BeforeValidator(check_center_of_mass)] = Field( + ..., + description=has_image_center_of_mass_center_of_mass_description, + ) + + channel: typing.Optional[StrictInt] = Field( + None, + description=has_image_center_of_mass_channel_description, + ) + + eps: Annotated[typing.Union[StrictInt, StrictFloat], BeforeValidator(check_non_negative_if_set)] = Field( + 0.01, + description=has_image_center_of_mass_eps_description, + ) + + +has_image_channels_channels_description = """Expected number of channels of the image.""" + +has_image_channels_delta_description = """Maximum allowed difference of the number of channels (default is 0). The observed number of channels has to be in the range ``value +- delta``.""" + +has_image_channels_min_description = """Minimum allowed number of channels.""" + +has_image_channels_max_description = """Maximum allowed number of channels.""" + +has_image_channels_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_image_channels_model(AssertionModel): + r"""Asserts the output is an image and has a specific number of channels. + + The number of channels is plus/minus ``delta`` (e.g., ````). + + Alternatively the range of the expected number of channels can be specified by ``min`` and/or ``max``.""" + + that: Literal["has_image_channels"] = "has_image_channels" + + channels: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_channels_channels_description, + ) + + delta: Annotated[StrictInt, BeforeValidator(check_non_negative_if_set)] = Field( + 0, + description=has_image_channels_delta_description, + ) + + min: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_channels_min_description, + ) + + max: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_channels_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_image_channels_negate_description, + ) + + +has_image_height_height_description = """Expected height of the image (in pixels).""" + +has_image_height_delta_description = """Maximum allowed difference of the image height (in pixels, default is 0). The observed height has to be in the range ``value +- delta``.""" + +has_image_height_min_description = """Minimum allowed height of the image (in pixels).""" + +has_image_height_max_description = """Maximum allowed height of the image (in pixels).""" + +has_image_height_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_image_height_model(AssertionModel): + r"""Asserts the output is an image and has a specific height (in pixels). + + The height is plus/minus ``delta`` (e.g., ````). + Alternatively the range of the expected height can be specified by ``min`` and/or ``max``.""" + + that: Literal["has_image_height"] = "has_image_height" + + height: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_height_height_description, + ) + + delta: Annotated[StrictInt, BeforeValidator(check_non_negative_if_set)] = Field( + 0, + description=has_image_height_delta_description, + ) + + min: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_height_min_description, + ) + + max: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_height_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_image_height_negate_description, + ) + + +has_image_mean_intensity_channel_description = """Restricts the assertion to a specific channel of the image (where ``0`` corresponds to the first image channel).""" + +has_image_mean_intensity_mean_intensity_description = """The required mean value of the image intensities.""" + +has_image_mean_intensity_eps_description = """The absolute tolerance to be used for ``value`` (defaults to ``0.01``). The observed mean value of the image intensities has to be in the range ``value +- eps``.""" + +has_image_mean_intensity_min_description = """A lower bound of the required mean value of the image intensities.""" + +has_image_mean_intensity_max_description = """An upper bound of the required mean value of the image intensities.""" + + +class has_image_mean_intensity_model(AssertionModel): + r"""Asserts the output is an image and has a specific mean intensity value. + + The mean intensity value is plus/minus ``eps`` (e.g., ````). + Alternatively the range of the expected mean intensity value can be specified by ``min`` and/or ``max``.""" + + that: Literal["has_image_mean_intensity"] = "has_image_mean_intensity" + + channel: typing.Optional[StrictInt] = Field( + None, + description=has_image_mean_intensity_channel_description, + ) + + mean_intensity: typing.Optional[typing.Union[StrictInt, StrictFloat]] = Field( + None, + description=has_image_mean_intensity_mean_intensity_description, + ) + + eps: Annotated[typing.Union[StrictInt, StrictFloat], BeforeValidator(check_non_negative_if_set)] = Field( + 0.01, + description=has_image_mean_intensity_eps_description, + ) + + min: typing.Optional[typing.Union[StrictInt, StrictFloat]] = Field( + None, + description=has_image_mean_intensity_min_description, + ) + + max: typing.Optional[typing.Union[StrictInt, StrictFloat]] = Field( + None, + description=has_image_mean_intensity_max_description, + ) + + +has_image_mean_object_size_channel_description = """Restricts the assertion to a specific channel of the image (where ``0`` corresponds to the first image channel).""" + +has_image_mean_object_size_labels_description = """List of labels, separated by a comma. Labels *not* on this list will be excluded from consideration. Cannot be used in combination with ``exclude_labels``.""" + +has_image_mean_object_size_exclude_labels_description = """List of labels to be excluded from consideration, separated by a comma. The primary usage of this attribute is to exclude the background of a label image. Cannot be used in combination with ``labels``.""" + +has_image_mean_object_size_mean_object_size_description = """The required mean size of the uniquely labeled objects.""" + +has_image_mean_object_size_eps_description = """The absolute tolerance to be used for ``value`` (defaults to ``0.01``). The observed mean size of the uniquely labeled objects has to be in the range ``value +- eps``.""" + +has_image_mean_object_size_min_description = ( + """A lower bound of the required mean size of the uniquely labeled objects.""" +) + +has_image_mean_object_size_max_description = ( + """An upper bound of the required mean size of the uniquely labeled objects.""" +) + + +class has_image_mean_object_size_model(AssertionModel): + r"""Asserts the output is an image with labeled objects which have the specified mean size (number of pixels), + + The mean size is plus/minus ``eps`` (e.g., ````). + + The labels must be unique.""" + + that: Literal["has_image_mean_object_size"] = "has_image_mean_object_size" + + channel: typing.Optional[StrictInt] = Field( + None, + description=has_image_mean_object_size_channel_description, + ) + + labels: typing.Optional[typing.List[int]] = Field( + None, + description=has_image_mean_object_size_labels_description, + ) + + exclude_labels: typing.Optional[typing.List[int]] = Field( + None, + description=has_image_mean_object_size_exclude_labels_description, + ) + + mean_object_size: Annotated[ + typing.Optional[typing.Union[StrictInt, StrictFloat]], BeforeValidator(check_non_negative_if_set) + ] = Field( + None, + description=has_image_mean_object_size_mean_object_size_description, + ) + + eps: Annotated[typing.Union[StrictInt, StrictFloat], BeforeValidator(check_non_negative_if_set)] = Field( + 0.01, + description=has_image_mean_object_size_eps_description, + ) + + min: Annotated[ + typing.Optional[typing.Union[StrictInt, StrictFloat]], BeforeValidator(check_non_negative_if_set) + ] = Field( + None, + description=has_image_mean_object_size_min_description, + ) + + max: Annotated[ + typing.Optional[typing.Union[StrictInt, StrictFloat]], BeforeValidator(check_non_negative_if_set) + ] = Field( + None, + description=has_image_mean_object_size_max_description, + ) + + +has_image_n_labels_channel_description = """Restricts the assertion to a specific channel of the image (where ``0`` corresponds to the first image channel).""" + +has_image_n_labels_labels_description = """List of labels, separated by a comma. Labels *not* on this list will be excluded from consideration. Cannot be used in combination with ``exclude_labels``.""" + +has_image_n_labels_exclude_labels_description = """List of labels to be excluded from consideration, separated by a comma. The primary usage of this attribute is to exclude the background of a label image. Cannot be used in combination with ``labels``.""" + +has_image_n_labels_n_description = """Expected number of labels.""" + +has_image_n_labels_delta_description = """Maximum allowed difference of the number of labels (default is 0). The observed number of labels has to be in the range ``value +- delta``.""" + +has_image_n_labels_min_description = """Minimum allowed number of labels.""" + +has_image_n_labels_max_description = """Maximum allowed number of labels.""" + +has_image_n_labels_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_image_n_labels_model(AssertionModel): + r"""Asserts the output is an image and has the specified labels. + + Labels can be a number of labels or unique values (e.g., + ````). + + The primary usage of this assertion is to verify the number of objects in images with uniquely labeled objects.""" + + that: Literal["has_image_n_labels"] = "has_image_n_labels" + + channel: typing.Optional[StrictInt] = Field( + None, + description=has_image_n_labels_channel_description, + ) + + labels: typing.Optional[typing.List[int]] = Field( + None, + description=has_image_n_labels_labels_description, + ) + + exclude_labels: typing.Optional[typing.List[int]] = Field( + None, + description=has_image_n_labels_exclude_labels_description, + ) + + n: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_n_labels_n_description, + ) + + delta: Annotated[StrictInt, BeforeValidator(check_non_negative_if_set)] = Field( + 0, + description=has_image_n_labels_delta_description, + ) + + min: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_n_labels_min_description, + ) + + max: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_n_labels_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_image_n_labels_negate_description, + ) + + +has_image_width_width_description = """Expected width of the image (in pixels).""" + +has_image_width_delta_description = """Maximum allowed difference of the image width (in pixels, default is 0). The observed width has to be in the range ``value +- delta``.""" + +has_image_width_min_description = """Minimum allowed width of the image (in pixels).""" + +has_image_width_max_description = """Maximum allowed width of the image (in pixels).""" + +has_image_width_negate_description = """A boolean that can be set to true to negate the outcome of the assertion.""" + + +class has_image_width_model(AssertionModel): + r"""Asserts the output is an image and has a specific width (in pixels). + + The width is plus/minus ``delta`` (e.g., ````). + Alternatively the range of the expected width can be specified by ``min`` and/or ``max``.""" + + that: Literal["has_image_width"] = "has_image_width" + + width: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_width_width_description, + ) + + delta: Annotated[StrictInt, BeforeValidator(check_non_negative_if_set)] = Field( + 0, + description=has_image_width_delta_description, + ) + + min: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_width_min_description, + ) + + max: Annotated[typing.Optional[StrictInt], BeforeValidator(check_non_negative_if_set)] = Field( + None, + description=has_image_width_max_description, + ) + + negate: typing.Union[bool, str] = Field( + False, + description=has_image_width_negate_description, + ) + + +any_assertion_model = Annotated[ + typing.Union[ + has_line_model, + has_line_matching_model, + has_n_lines_model, + has_text_model, + has_text_matching_model, + not_has_text_model, + has_n_columns_model, + attribute_is_model, + attribute_matches_model, + element_text_model, + element_text_is_model, + element_text_matches_model, + has_element_with_path_model, + has_n_elements_with_path_model, + is_valid_xml_model, + xml_element_model, + has_json_property_with_text_model, + has_json_property_with_value_model, + has_h5_attribute_model, + has_h5_keys_model, + has_archive_member_model, + has_size_model, + has_image_center_of_mass_model, + has_image_channels_model, + has_image_height_model, + has_image_mean_intensity_model, + has_image_mean_object_size_model, + has_image_n_labels_model, + has_image_width_model, + ], + Field(discriminator="that"), +] + +assertion_list = RootModel[typing.List[any_assertion_model]] diff --git a/lib/galaxy/tool_util/verify/asserts/__init__.py b/lib/galaxy/tool_util/verify/asserts/__init__.py index 10e0ce8909e4..532692c6198a 100644 --- a/lib/galaxy/tool_util/verify/asserts/__init__.py +++ b/lib/galaxy/tool_util/verify/asserts/__init__.py @@ -5,6 +5,11 @@ getmembers, ) from tempfile import NamedTemporaryFile +from typing import ( + Callable, + Dict, + Tuple, +) from galaxy.util import unicodify from galaxy.util.compression_utils import get_fileobj @@ -13,11 +18,8 @@ assertion_module_names = ["text", "tabular", "xml", "json", "hdf5", "archive", "size", "image"] -# Code for loading modules containing assertion checking functions, to -# create a new module of assertion functions, create the needed python -# source file "test/base/asserts/.py" and add -# to the list of assertion module names defined above. -assertion_functions = {} +assertion_module_and_functions: Dict[str, Tuple[str, Callable]] = {} + for assertion_module_name in assertion_module_names: full_assertion_module_name = f"galaxy.tool_util.verify.asserts.{assertion_module_name}" try: @@ -29,10 +31,17 @@ continue for member, value in getmembers(assertion_module): if member.startswith("assert_"): - assertion_functions[member] = value + assertion_module_and_functions[member] = (f"{full_assertion_module_name}.{member}", value) + + +# Code for loading modules containing assertion checking functions, to +# create a new module of assertion functions, create the needed python +# source file "test/base/asserts/.py" and add +# to the list of assertion module names defined above. +assertion_functions: Dict[str, Callable] = {k: v[1] for (k, v) in assertion_module_and_functions.items()} -def verify_assertions(data: bytes, assertion_description_list, decompress: bool = False): +def verify_assertions(data: bytes, assertion_description_list: list, decompress: bool = False): """This function takes a list of assertions and a string to check these assertions against.""" if decompress: diff --git a/lib/galaxy/tool_util/verify/asserts/_types.py b/lib/galaxy/tool_util/verify/asserts/_types.py new file mode 100644 index 000000000000..4b2f601f9d5a --- /dev/null +++ b/lib/galaxy/tool_util/verify/asserts/_types.py @@ -0,0 +1,102 @@ +from typing import ( + Any, + List, + Optional, + Union, +) + +from typing_extensions import ( + Annotated, + Protocol, +) + + +class AssertionParameter: + doc: str + xml_type: Optional[str] + json_type: Optional[str] + deprecated: bool + validators: List[str] + + def __init__( + self, + doc: Optional[str], + xml_type: Optional[str] = None, + json_type: Optional[str] = None, + deprecated: bool = False, + validators: Optional[List[str]] = None, + ): + self.doc = doc or "" + self.xml_type = xml_type + self.json_type = json_type + self.deprecated = deprecated + self.validators = validators or [] + + +XmlInt = Union[int, str] +XmlFloat = Union[float, str] +XmlBool = Union[bool, str] +XmlRegex = str +OptionalXmlInt = Optional[XmlInt] +OptionalXmlFloat = Optional[XmlFloat] +OptionalXmlBool = Optional[XmlBool] + +Output = Annotated[str, "The target output of a tool or workflow read as a UTF-8 string"] +OutputBytes = Annotated[bytes, "The target output of a tool or workflow read as raw Python 'bytes'"] + + +class VerifyAssertionsFunction(Protocol): + + def __call__(self, data: bytes, assertion_description_list: list, decompress: bool = False): + """Callback for recursirve functions.""" + + +ChildAssertions = Annotated[Any, "Parsed child assertions"] +Negate = Annotated[ + XmlBool, + AssertionParameter( + "A boolean that can be set to true to negate the outcome of the assertion.", xml_type="PermissiveBoolean" + ), +] +NEGATE_DEFAULT = False + +N = Annotated[ + Optional[XmlInt], AssertionParameter("Desired number, can be suffixed by ``(k|M|G|T|P|E)i?``", xml_type="Bytes") +] +Delta = Annotated[ + XmlInt, + AssertionParameter( + "Allowed difference with respect to n (default: 0), can be suffixed by ``(k|M|G|T|P|E)i?``", xml_type="Bytes" + ), +] +Min = Annotated[ + Optional[XmlInt], + AssertionParameter("Minimum number (default: -infinity), can be suffixed by ``(k|M|G|T|P|E)i?``", xml_type="Bytes"), +] +Max = Annotated[ + Optional[XmlInt], + AssertionParameter("Maximum number (default: infinity), can be suffixed by ``(k|M|G|T|P|E)i?``", xml_type="Bytes"), +] + + +__all__ = ( + "Annotated", + "AssertionParameter", + "ChildAssertions", + "Delta", + "Max", + "Min", + "Negate", + "N", + "NEGATE_DEFAULT", + "OptionalXmlBool", + "OptionalXmlFloat", + "OptionalXmlInt", + "Output", + "OutputBytes", + "VerifyAssertionsFunction", + "XmlBool", + "XmlFloat", + "XmlInt", + "XmlRegex", +) diff --git a/lib/galaxy/tool_util/verify/asserts/archive.py b/lib/galaxy/tool_util/verify/asserts/archive.py index 336b3ccfaee6..6bf554c10fd0 100644 --- a/lib/galaxy/tool_util/verify/asserts/archive.py +++ b/lib/galaxy/tool_util/verify/asserts/archive.py @@ -3,12 +3,22 @@ import tarfile import tempfile import zipfile -from typing import ( - Optional, - Union, -) from galaxy.util import asbool +from ._types import ( + Annotated, + AssertionParameter, + ChildAssertions, + Delta, + Max, + Min, + N, + Negate, + NEGATE_DEFAULT, + OutputBytes, + VerifyAssertionsFunction, + XmlBool, +) from ._util import _assert_presence_number @@ -55,21 +65,72 @@ def _list_from_zip(output_bytes, path): return sorted(lst) +Path = Annotated[str, AssertionParameter("The regular expression specifying the archive member.")] +All = Annotated[ + XmlBool, + AssertionParameter( + "Check the sub-assertions for all paths matching the path. Default: false, i.e. only the first", + xml_type="PermissiveBoolean", + ), +] + + def assert_has_archive_member( - output_bytes: bytes, - path: str, - verify_assertions_function, - children, - all: Union[bool, str] = False, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output_bytes: OutputBytes, + path: Path, + verify_assertions_function: VerifyAssertionsFunction, + children: ChildAssertions = None, + all: All = False, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """Recursively checks the specified children assertions against the text of - the first element matching the specified path found within the archive. - Currently supported formats: .zip, .tar, .tar.gz.""" + """This tag allows to check if ``path`` is contained in a compressed file. + + The path is a regular expression that is matched against the full paths of the objects in + the compressed file (remember that "matching" means it is checked if a prefix of + the full path of an archive member is described by the regular expression). + Valid archive formats include ``.zip``, ``.tar``, and ``.tar.gz``. Note that + depending on the archive creation method: + + - full paths of the members may be prefixed with ``./`` + - directories may be treated as empty files + + ```xml + + ``` + + With ``n`` and ``delta`` (or ``min`` and ``max``) assertions on the number of + archive members matching ``path`` can be expressed. The following could be used, + e.g., to assert an archive containing n±1 elements out of which at least + 4 need to have a ``txt`` extension. + + ```xml + + + ``` + + In addition the tag can contain additional assertions as child elements about + the first member in the archive matching the regular expression ``path``. For + instance + + ```xml + + + + ``` + + If the ``all`` attribute is set to ``true`` then all archive members are subject + to the assertions. Note that, archive members matching the ``path`` are sorted + alphabetically. + + The ``negate`` attribute of the ``has_archive_member`` assertion only affects + the asserts on the presence and number of matching archive members, but not any + sub-assertions (which can offer the ``negate`` attribute on their own). The + check if the file is an archive at all, which is also done by the function, is + not affected.""" all = asbool(all) extract_foo = None # from python 3.9 is_tarfile supports file like objects then we do not need diff --git a/lib/galaxy/tool_util/verify/asserts/hdf5.py b/lib/galaxy/tool_util/verify/asserts/hdf5.py index ddbbcdd35cda..2e44e0f17f43 100644 --- a/lib/galaxy/tool_util/verify/asserts/hdf5.py +++ b/lib/galaxy/tool_util/verify/asserts/hdf5.py @@ -5,17 +5,31 @@ except ImportError: h5py = None +from ._types import ( + Annotated, + AssertionParameter, + OutputBytes, +) + IMPORT_MISSING_MESSAGE = "h5 assertion requires unavailable optional dependency h5py" +Key = Annotated[str, AssertionParameter("HDF5 attribute to check value of.")] +Value = Annotated[str, AssertionParameter("Expected value of HDF5 attribute to check.")] +Keys = Annotated[str, AssertionParameter("HDF5 attributes to check value of as a comma-separated string.")] + def _assert_h5py(): if h5py is None: raise Exception(IMPORT_MISSING_MESSAGE) -def assert_has_h5_attribute(output_bytes: bytes, key: str, value: str) -> None: - """Asserts the specified HDF5 output has a given key-value pair as HDF5 - attribute""" +def assert_has_h5_attribute(output_bytes: OutputBytes, key: Key, value: Value) -> None: + """Asserts HDF5 output contains the specified ``value`` for an attribute (``key``), e.g. + + ```xml + + ``` + """ _assert_h5py() output_temp = io.BytesIO(output_bytes) local_attrs = h5py.File(output_temp, "r").attrs @@ -25,7 +39,7 @@ def assert_has_h5_attribute(output_bytes: bytes, key: str, value: str) -> None: # TODO the function actually queries groups. so the function and argument name are misleading -def assert_has_h5_keys(output_bytes: bytes, keys: str) -> None: +def assert_has_h5_keys(output_bytes: OutputBytes, keys: Keys) -> None: """Asserts the specified HDF5 output has the given keys.""" _assert_h5py() h5_keys = sorted([k.strip() for k in keys.strip().split(",")]) diff --git a/lib/galaxy/tool_util/verify/asserts/image.py b/lib/galaxy/tool_util/verify/asserts/image.py index 7513be700f5e..c56bd144bc61 100644 --- a/lib/galaxy/tool_util/verify/asserts/image.py +++ b/lib/galaxy/tool_util/verify/asserts/image.py @@ -8,6 +8,17 @@ Union, ) +from ._types import ( + Annotated, + AssertionParameter, + Negate, + NEGATE_DEFAULT, + OptionalXmlFloat, + OptionalXmlInt, + OutputBytes, + XmlFloat, + XmlInt, +) from ._util import _assert_number try: @@ -27,6 +38,255 @@ import numpy.typing +JSON_STRICT_NUMBER = "typing.Union[StrictInt, StrictFloat]" +JSON_OPTIONAL_STRICT_NUMBER = f"typing.Optional[{JSON_STRICT_NUMBER}]" + + +Width = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Expected width of the image (in pixels).", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +Height = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Expected height of the image (in pixels).", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +Channels = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Expected number of channels of the image.", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +WidthDelta = Annotated[ + XmlInt, + AssertionParameter( + "Maximum allowed difference of the image width (in pixels, default is 0). The observed width has to be in the range ``value +- delta``.", + json_type="StrictInt", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +WidthMin = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Minimum allowed width of the image (in pixels).", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +WidthMax = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Maximum allowed width of the image (in pixels).", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +HeightDelta = Annotated[ + XmlInt, + AssertionParameter( + "Maximum allowed difference of the image height (in pixels, default is 0). The observed height has to be in the range ``value +- delta``.", + json_type="StrictInt", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +HeightMin = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Minimum allowed height of the image (in pixels).", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +HeightMax = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Maximum allowed height of the image (in pixels).", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +ChannelsDelta = Annotated[ + XmlInt, + AssertionParameter( + "Maximum allowed difference of the number of channels (default is 0). The observed number of channels has to be in the range ``value +- delta``.", + json_type="StrictInt", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +ChannelsMin = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Minimum allowed number of channels.", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +ChannelsMax = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Maximum allowed number of channels.", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +MeanIntensity = Annotated[ + OptionalXmlFloat, + AssertionParameter("The required mean value of the image intensities.", json_type=JSON_OPTIONAL_STRICT_NUMBER), +] +MeanIntensityEps = Annotated[ + XmlFloat, + AssertionParameter( + "The absolute tolerance to be used for ``value`` (defaults to ``0.01``). The observed mean value of the image intensities has to be in the range ``value +- eps``.", + json_type=JSON_STRICT_NUMBER, + validators=["check_non_negative_if_set"], + ), +] +MeanIntensityMin = Annotated[ + OptionalXmlFloat, + AssertionParameter( + "A lower bound of the required mean value of the image intensities.", json_type=JSON_OPTIONAL_STRICT_NUMBER + ), +] +MeanIntensityMax = Annotated[ + OptionalXmlFloat, + AssertionParameter( + "An upper bound of the required mean value of the image intensities.", json_type=JSON_OPTIONAL_STRICT_NUMBER + ), +] +NumLabels = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Expected number of labels.", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +NumLabelsDelta = Annotated[ + XmlInt, + AssertionParameter( + "Maximum allowed difference of the number of labels (default is 0). The observed number of labels has to be in the range ``value +- delta``.", + json_type="StrictInt", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +NumLabelsMin = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Minimum allowed number of labels.", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] +NumLabelsMax = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Maximum allowed number of labels.", + json_type="typing.Optional[StrictInt]", + xml_type="xs:nonNegativeInteger", + validators=["check_non_negative_if_set"], + ), +] + +Channel = Annotated[ + OptionalXmlInt, + AssertionParameter( + "Restricts the assertion to a specific channel of the image (where ``0`` corresponds to the first image channel).", + json_type="typing.Optional[StrictInt]", + ), +] +CenterOfMass = Annotated[ + str, + AssertionParameter( + "The required center of mass of the image intensities (horizontal and vertical coordinate, separated by a comma).", + validators=["check_center_of_mass"], + ), +] +CenterOfMassEps = Annotated[ + XmlFloat, + AssertionParameter( + "The maximum allowed Euclidean distance to the required center of mass (defaults to ``0.01``).", + json_type=JSON_STRICT_NUMBER, + validators=["check_non_negative_if_set"], + ), +] +Labels = Annotated[ + Optional[Union[str, List[int]]], + AssertionParameter( + "List of labels, separated by a comma. Labels *not* on this list will be excluded from consideration. Cannot be used in combination with ``exclude_labels``.", + xml_type="xs:string", + json_type="typing.Optional[typing.List[int]]", + ), +] +ExcludeLabels = Annotated[ + Optional[Union[str, List[int]]], + AssertionParameter( + "List of labels to be excluded from consideration, separated by a comma. The primary usage of this attribute is to exclude the background of a label image. Cannot be used in combination with ``labels``.", + xml_type="xs:string", + json_type="typing.Optional[typing.List[int]]", + ), +] +MeanObjectSize = Annotated[ + OptionalXmlFloat, + AssertionParameter( + "The required mean size of the uniquely labeled objects.", + xml_type="xs:float", + json_type=JSON_OPTIONAL_STRICT_NUMBER, + validators=["check_non_negative_if_set"], + ), +] +MeanObjectSizeEps = Annotated[ + XmlFloat, + AssertionParameter( + "The absolute tolerance to be used for ``value`` (defaults to ``0.01``). The observed mean size of the uniquely labeled objects has to be in the range ``value +- eps``.", + xml_type="xs:float", + json_type=JSON_STRICT_NUMBER, + validators=["check_non_negative_if_set"], + ), +] +MeanObjectSizeMin = Annotated[ + OptionalXmlFloat, + AssertionParameter( + "A lower bound of the required mean size of the uniquely labeled objects.", + xml_type="xs:float", + json_type=JSON_OPTIONAL_STRICT_NUMBER, + validators=["check_non_negative_if_set"], + ), +] +MeanObjectSizeMax = Annotated[ + OptionalXmlFloat, + AssertionParameter( + "An upper bound of the required mean size of the uniquely labeled objects.", + xml_type="xs:float", + json_type=JSON_OPTIONAL_STRICT_NUMBER, + validators=["check_non_negative_if_set"], + ), +] + + def _assert_float( actual: float, label: str, @@ -52,15 +312,17 @@ def _assert_float( def assert_has_image_width( - output_bytes: bytes, - width: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output_bytes: OutputBytes, + width: Width = None, + delta: WidthDelta = 0, + min: WidthMin = None, + max: WidthMax = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """ - Asserts the specified output is an image and has a width of the specified value. + """Asserts the output is an image and has a specific width (in pixels). + + The width is plus/minus ``delta`` (e.g., ````). + Alternatively the range of the expected width can be specified by ``min`` and/or ``max``. """ im_arr = _get_image(output_bytes) _assert_number( @@ -77,14 +339,16 @@ def assert_has_image_width( def assert_has_image_height( output_bytes: bytes, - height: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + height: Height = None, + delta: HeightDelta = 0, + min: HeightMin = None, + max: HeightMax = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """ - Asserts the specified output is an image and has a height of the specified value. + """Asserts the output is an image and has a specific height (in pixels). + + The height is plus/minus ``delta`` (e.g., ````). + Alternatively the range of the expected height can be specified by ``min`` and/or ``max``. """ im_arr = _get_image(output_bytes) _assert_number( @@ -101,14 +365,17 @@ def assert_has_image_height( def assert_has_image_channels( output_bytes: bytes, - channels: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + channels: Channels = None, + delta: ChannelsDelta = 0, + min: ChannelsMin = None, + max: ChannelsMax = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """ - Asserts the specified output is an image and has the specified number of channels. + """Asserts the output is an image and has a specific number of channels. + + The number of channels is plus/minus ``delta`` (e.g., ````). + + Alternatively the range of the expected number of channels can be specified by ``min`` and/or ``max``. """ im_arr = _get_image(output_bytes) n_channels = 1 if im_arr.ndim < 3 else im_arr.shape[2] # we assume here that the image is a 2-D image @@ -165,15 +432,17 @@ def _get_image( def assert_has_image_mean_intensity( - output_bytes: bytes, - channel: Optional[Union[int, str]] = None, - mean_intensity: Optional[Union[float, str]] = None, - eps: Union[float, str] = 0.01, - min: Optional[Union[float, str]] = None, - max: Optional[Union[float, str]] = None, + output_bytes: OutputBytes, + channel: Channel = None, + mean_intensity: MeanIntensity = None, + eps: MeanIntensityEps = 0.01, + min: MeanIntensityMin = None, + max: MeanIntensityMax = None, ) -> None: - """ - Asserts the specified output is an image and has the specified mean intensity value. + """Asserts the output is an image and has a specific mean intensity value. + + The mean intensity value is plus/minus ``eps`` (e.g., ````). + Alternatively the range of the expected mean intensity value can be specified by ``min`` and/or ``max``. """ im_arr = _get_image(output_bytes, channel) _assert_float( @@ -187,22 +456,24 @@ def assert_has_image_mean_intensity( def assert_has_image_center_of_mass( - output_bytes: bytes, - center_of_mass: Union[Tuple[float, float], str], - channel: Optional[Union[int, str]] = None, - eps: Union[float, str] = 0.01, + output_bytes: OutputBytes, + center_of_mass: CenterOfMass, + channel: Channel = None, + eps: CenterOfMassEps = 0.01, ) -> None: - """ - Asserts the specified output is an image and has the specified center of mass. + """Asserts the specified output is an image and has the specified center of mass. + + Asserts the output is an image and has a specific center of mass, + or has an Euclidean distance of ``eps`` or less to that point (e.g., + ````). """ im_arr = _get_image(output_bytes, channel) - if isinstance(center_of_mass, str): - center_of_mass_parts = [c.strip() for c in center_of_mass.split(",")] - assert len(center_of_mass_parts) == 2 - center_of_mass = (float(center_of_mass_parts[0]), float(center_of_mass_parts[1])) - assert len(center_of_mass) == 2, "center_of_mass must have two components" + center_of_mass_parts = [c.strip() for c in center_of_mass.split(",")] + assert len(center_of_mass_parts) == 2 + center_of_mass_tuple = (float(center_of_mass_parts[0]), float(center_of_mass_parts[1])) + assert len(center_of_mass_tuple) == 2, "center_of_mass must have two components" actual_center_of_mass = _compute_center_of_mass(im_arr) - distance = numpy.linalg.norm(numpy.subtract(center_of_mass, actual_center_of_mass)) + distance = numpy.linalg.norm(numpy.subtract(center_of_mass_tuple, actual_center_of_mass)) assert distance <= float( eps ), f"Wrong center of mass: {actual_center_of_mass} (expected {center_of_mass}, distance: {distance}, eps: {eps})" @@ -251,18 +522,22 @@ def cast_label(label): def assert_has_image_n_labels( - output_bytes: bytes, - channel: Optional[Union[int, str]] = None, - labels: Optional[Union[str, List[int]]] = None, - exclude_labels: Optional[Union[str, List[int]]] = None, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output_bytes: OutputBytes, + channel: Channel = None, + labels: Labels = None, + exclude_labels: ExcludeLabels = None, + n: NumLabels = None, + delta: NumLabelsDelta = 0, + min: NumLabelsMin = None, + max: NumLabelsMax = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """ - Asserts the specified output is an image and has the specified number of unique values (e.g., uniquely labeled objects). + """Asserts the output is an image and has the specified labels. + + Labels can be a number of labels or unique values (e.g., + ````). + + The primary usage of this assertion is to verify the number of objects in images with uniquely labeled objects. """ present_labels = _get_image_labels(output_bytes, channel, labels, exclude_labels)[1] _assert_number( @@ -278,17 +553,20 @@ def assert_has_image_n_labels( def assert_has_image_mean_object_size( - output_bytes: bytes, - channel: Optional[Union[int, str]] = None, - labels: Optional[Union[str, List[int]]] = None, - exclude_labels: Optional[Union[str, List[int]]] = None, - mean_object_size: Optional[Union[float, str]] = None, - eps: Union[float, str] = 0.01, - min: Optional[Union[float, str]] = None, - max: Optional[Union[float, str]] = None, + output_bytes: OutputBytes, + channel: Channel = None, + labels: Labels = None, + exclude_labels: ExcludeLabels = None, + mean_object_size: MeanObjectSize = None, + eps: MeanObjectSizeEps = 0.01, + min: MeanObjectSizeMin = None, + max: MeanObjectSizeMax = None, ) -> None: - """ - Asserts the specified output is an image with labeled objects which have the specified mean size (number of pixels). + """Asserts the output is an image with labeled objects which have the specified mean size (number of pixels), + + The mean size is plus/minus ``eps`` (e.g., ````). + + The labels must be unique. """ im_arr, present_labels = _get_image_labels(output_bytes, channel, labels, exclude_labels) actual_mean_object_size = sum((im_arr == label).sum() for label in present_labels) / len(present_labels) diff --git a/lib/galaxy/tool_util/verify/asserts/json.py b/lib/galaxy/tool_util/verify/asserts/json.py index 9b475f35d980..1f67e6b46087 100644 --- a/lib/galaxy/tool_util/verify/asserts/json.py +++ b/lib/galaxy/tool_util/verify/asserts/json.py @@ -5,8 +5,20 @@ cast, ) +from ._types import ( + Annotated, + AssertionParameter, + Output, +) + PropertyVisitor = Callable[[str, Any], Any] +Property = Annotated[str, AssertionParameter("The property name to search the JSON document for.")] +Text = Annotated[str, AssertionParameter("The expected text value of the target JSON attribute.")] +Value = Annotated[ + str, AssertionParameter("The expected JSON value of the target JSON attribute (as a JSON encoded string).") +] + def any_in_tree(f: PropertyVisitor, json_tree: Any): if isinstance(json_tree, list): @@ -25,13 +37,18 @@ def any_in_tree(f: PropertyVisitor, json_tree: Any): def assert_has_json_property_with_value( - output, - property: str, - value: str, + output: Output, + property: Property, + value: Value, ): - """Assert JSON tree contains the specified property with specified JSON-ified value.""" - output_json = assert_json_and_load(output) - expected_value = assert_json_and_load(value) + """Asserts the JSON document contains a property or key with the specified JSON value. + + ```xml + + ``` + """ + output_json = _assert_json_and_load(output) + expected_value = _assert_json_and_load(value) def is_property(key, value): return key == property and value == expected_value @@ -40,12 +57,17 @@ def is_property(key, value): def assert_has_json_property_with_text( - output, - property: str, - text: str, + output: Output, + property: Property, + text: Text, ): - """Assert JSON tree contains the specified property with specified JSON-ified value.""" - output_json = assert_json_and_load(output) + """Asserts the JSON document contains a property or key with the specified text (i.e. string) value. + + ```xml + + ``` + """ + output_json = _assert_json_and_load(output) def is_property(key, value): return key == property and value == text @@ -53,7 +75,7 @@ def is_property(key, value): assert any_in_tree(is_property, output_json), f"Failed to find property [{property}] with text [{text}]" -def assert_json_and_load(json_str: str): +def _assert_json_and_load(json_str: str): try: return json.loads(json_str) except Exception: diff --git a/lib/galaxy/tool_util/verify/asserts/size.py b/lib/galaxy/tool_util/verify/asserts/size.py index 6eb6cb43094d..e4b3e8a6ef1f 100644 --- a/lib/galaxy/tool_util/verify/asserts/size.py +++ b/lib/galaxy/tool_util/verify/asserts/size.py @@ -1,24 +1,37 @@ -from typing import ( - Optional, - Union, +from ._types import ( + Annotated, + AssertionParameter, + Delta, + Max, + Min, + Negate, + NEGATE_DEFAULT, + OptionalXmlInt, + OutputBytes, ) - from ._util import _assert_number def assert_has_size( - output_bytes: bytes, - value: Optional[Union[int, str]] = None, - size: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output_bytes: OutputBytes, + value: Annotated[ + OptionalXmlInt, AssertionParameter("Deprecated alias for `size`", xml_type="Bytes", deprecated=True) + ] = None, + size: Annotated[ + OptionalXmlInt, + AssertionParameter( + "Desired size of the output (in bytes), can be suffixed by ``(k|M|G|T|P|E)i?``", xml_type="Bytes" + ), + ] = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """ - Asserts the specified output has a size of the specified value - (size and value or synonyms), - allowing for absolute (delta) and relative (delta_frac) difference. + """Asserts the specified output has a size of the specified value + + Attributes size and value or synonyms though value is considered deprecated. + The size optionally allows for absolute (``delta``) difference. """ output_size = len(output_bytes) if size is None: diff --git a/lib/galaxy/tool_util/verify/asserts/tabular.py b/lib/galaxy/tool_util/verify/asserts/tabular.py index b7278ec1d979..9fc8e9f71c05 100644 --- a/lib/galaxy/tool_util/verify/asserts/tabular.py +++ b/lib/galaxy/tool_util/verify/asserts/tabular.py @@ -1,11 +1,26 @@ import re -from typing import ( - Optional, - Union, -) +from ._types import ( + Annotated, + AssertionParameter, + Delta, + Max, + Min, + N, + Negate, + NEGATE_DEFAULT, + Output, +) from ._util import _assert_number +Sep = Annotated[str, AssertionParameter("Separator defining columns, default: tab")] +Comment = Annotated[ + str, + AssertionParameter( + "Comment character(s) used to skip comment lines (which should not be used for counting columns)" + ), +] + def get_first_line(output: str, comment: str) -> str: """ @@ -22,19 +37,26 @@ def get_first_line(output: str, comment: str) -> str: def assert_has_n_columns( - output: str, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - sep: str = "\t", - comment: str = "", - negate: Union[bool, str] = False, + output: Output, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + sep: Sep = "\t", + comment: Comment = "", + negate: Negate = NEGATE_DEFAULT, ) -> None: - """Asserts the tabular output contains n columns. The optional - sep argument specifies the column seperator used to determine the - number of columns. The optional comment argument specifies - comment characters""" + """Asserts tabular output contains the specified + number (``n``) of columns. + + For instance, ````. The assertion tests only the first line. + Number of columns can optionally also be specified with ``delta``. Alternatively the + range of expected occurences can be specified by ``min`` and/or ``max``. + + Optionally a column separator (``sep``, default is ``\t``) `and comment character(s) + can be specified (``comment``, default is empty string). The first non-comment + line is used for determining the number of columns. + """ first_line = get_first_line(output, comment) n_columns = len(first_line.split(sep)) _assert_number( diff --git a/lib/galaxy/tool_util/verify/asserts/text.py b/lib/galaxy/tool_util/verify/asserts/text.py index e7aa1bc37e4d..8726535ce999 100644 --- a/lib/galaxy/tool_util/verify/asserts/text.py +++ b/lib/galaxy/tool_util/verify/asserts/text.py @@ -1,23 +1,35 @@ import re -from typing import ( - Optional, - Union, -) +from typing_extensions import Annotated + +from ._types import ( + AssertionParameter, + Delta, + Max, + Min, + N, + Negate, + NEGATE_DEFAULT, + Output, +) from ._util import ( _assert_number, _assert_presence_number, ) +Text = Annotated[str, AssertionParameter("The text to search for in the output.")] +Line = Annotated[str, AssertionParameter("The full line of text to search for in the output.")] +Expression = Annotated[str, AssertionParameter("The regular expressions to attempt match in the output.")] + def assert_has_text( - output: str, - text: str, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output: Output, + text: Text, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: """Asserts specified output contains the substring specified by the argument text. The exact number of occurrences can be @@ -39,7 +51,7 @@ def assert_has_text( ) -def assert_not_has_text(output: str, text: str) -> None: +def assert_not_has_text(output: Output, text: Text) -> None: """Asserts specified output does not contain the substring specified by the argument text""" assert output is not None, "Checking not_has_text assertion on empty output (None)" @@ -47,13 +59,13 @@ def assert_not_has_text(output: str, text: str) -> None: def assert_has_line( - output: str, - line: str, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output: Output, + line: Line, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: """Asserts the specified output contains the line specified by the argument line. The exact number of occurrences can be optionally @@ -76,12 +88,12 @@ def assert_has_line( def assert_has_n_lines( - output: str, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output: Output, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: """Asserts the specified output contains ``n`` lines allowing for a difference in the number of lines (delta) @@ -101,13 +113,13 @@ def assert_has_n_lines( def assert_has_text_matching( - output: str, - expression: str, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output: Output, + expression: Expression, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: """Asserts the specified output contains text matching the regular expression specified by the argument expression. @@ -131,13 +143,13 @@ def assert_has_text_matching( def assert_has_line_matching( - output: str, - expression: str, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output: Output, + expression: Expression, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: """Asserts the specified output contains a line matching the regular expression specified by the argument expression. If n is given diff --git a/lib/galaxy/tool_util/verify/asserts/xml.py b/lib/galaxy/tool_util/verify/asserts/xml.py index b80798436e2e..848602f821ce 100644 --- a/lib/galaxy/tool_util/verify/asserts/xml.py +++ b/lib/galaxy/tool_util/verify/asserts/xml.py @@ -1,8 +1,5 @@ import re -from typing import ( - Optional, - Union, -) +from typing import Optional from lxml.etree import XMLSyntaxError @@ -12,95 +9,247 @@ parse_xml_string, unicodify, ) +from ._types import ( + Annotated, + AssertionParameter, + ChildAssertions, + Delta, + Max, + Min, + N, + Negate, + NEGATE_DEFAULT, + Output, + VerifyAssertionsFunction, + XmlBool, + XmlRegex, +) + +Path = Annotated[str, AssertionParameter("The Python xpath-like expression to find the target element.")] +ElementExpression = Annotated[ + XmlRegex, + AssertionParameter("The regular expressions to apply against the target element.", validators=["check_regex"]), +] +AttributeExpression = Annotated[ + XmlRegex, + AssertionParameter( + "The regular expressions to apply against the named attribute on the target XML element.", + validators=["check_regex"], + ), +] +Attribute = Annotated[str, AssertionParameter("The XML attribute name to test against from the target XML element.")] +OptionalAttribute = Annotated[ + Optional[str], AssertionParameter("The XML attribute name to test against from the target XML element.") +] +ElementText = Annotated[ + str, AssertionParameter("The expected element text (body of the XML tag) to test against on the target XML element") +] +AttributeText = Annotated[ + str, AssertionParameter("The expected attribute value to test against on the target XML element") +] +All = Annotated[ + XmlBool, + AssertionParameter( + "Check the sub-assertions for all paths matching the path. Default: false, i.e. only the first ", + xml_type="PermissiveBoolean", + ), +] -def assert_is_valid_xml(output: str) -> None: - """Simple assertion that just verifies the specified output - is valid XML.""" +def assert_is_valid_xml(output: Output) -> None: + """Asserts the output is a valid XML file (e.g. ````).""" try: parse_xml_string(output) except XMLSyntaxError as e: raise AssertionError(f"Expected valid XML, but could not parse output. {unicodify(e)}") -def assert_has_element_with_path(output: str, path: str, negate: Union[bool, str] = False) -> None: - """Asserts the specified output has at least one XML element with a - path matching the specified path argument. Valid paths are the - simplified subsets of XPath implemented by lxml.etree; - https://lxml.de/xpathxslt.html for more information.""" +def assert_has_element_with_path(output: Output, path: Path, negate: Negate = NEGATE_DEFAULT) -> None: + """Asserts the XML output contains at least one element (or tag) with the specified + XPath-like ``path``, e.g. + + ```xml + + ``` + + With ``negate`` the result of the assertion can be inverted.""" assert_xml_element(output, path, negate=negate) def assert_has_n_elements_with_path( - output: str, - path: str, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output: Output, + path: Path, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """Asserts the specified output has exactly n elements matching the - path specified.""" + """Asserts the XML output contains the specified number (``n``, optionally with ``delta``) of elements (or + tags) with the specified XPath-like ``path``. + + For example: + + ```xml + + ``` + + Alternatively to ``n`` and ``delta`` also the ``min`` and ``max`` attributes + can be used to specify the range of the expected number of occurences. + With ``negate`` the result of the assertion can be inverted. + """ assert_xml_element(output, path, n=n, delta=delta, min=min, max=max, negate=negate) -def assert_element_text_matches(output: str, path: str, expression: str, negate: Union[bool, str] = False) -> None: - """Asserts the text of the first element matching the specified - path matches the specified regular expression.""" +def assert_element_text_matches( + output: Output, path: Path, expression: ElementExpression, negate: Negate = NEGATE_DEFAULT +) -> None: + """Asserts the text of the XML element with the specified XPath-like ``path`` + matches the regular expression defined by ``expression``. + + For example: + + ```xml + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the assertion (on the matching) can be inverted (the + implicit assertion on the existence of the path is not affected). + """ sub = {"tag": "has_text_matching", "attributes": {"expression": expression, "negate": negate}} assert_xml_element(output, path, asserts.verify_assertions, [sub]) -def assert_element_text_is(output: str, path: str, text: str, negate: Union[bool, str] = False) -> None: - """Asserts the text of the first element matching the specified - path matches exactly the specified text.""" +def assert_element_text_is(output: Output, path: Path, text: ElementText, negate: Negate = NEGATE_DEFAULT) -> None: + """Asserts the text of the XML element with the specified XPath-like ``path`` is + the specified ``text``. + + For example: + + ```xml + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the assertion (on the equality) can be inverted (the + implicit assertion on the existence of the path is not affected). + """ assert_element_text_matches(output, path, re.escape(text) + "$", negate=negate) def assert_attribute_matches( - output: str, path: str, attribute, expression: str, negate: Union[bool, str] = False + output: Output, + path: Path, + attribute: Attribute, + expression: AttributeExpression, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """Asserts the specified attribute of the first element matching - the specified path matches the specified regular expression.""" + """Asserts the XML ``attribute`` for the element (or tag) with the specified + XPath-like ``path`` matches the regular expression specified by ``expression``. + + For example: + + ```xml + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the assertion (on the matching) can be inverted (the + implicit assertion on the existence of the path is not affected). + """ sub = {"tag": "has_text_matching", "attributes": {"expression": expression, "negate": negate}} assert_xml_element(output, path, asserts.verify_assertions, [sub], attribute=attribute) -def assert_attribute_is(output: str, path: str, attribute: str, text, negate: Union[bool, str] = False) -> None: - """Asserts the specified attribute of the first element matching - the specified path matches exactly the specified text.""" +def assert_attribute_is( + output: Output, path: Path, attribute: Attribute, text: AttributeText, negate: Negate = NEGATE_DEFAULT +) -> None: + """Asserts the XML ``attribute`` for the element (or tag) with the specified + XPath-like ``path`` is the specified ``text``. + + For example: + + ```xml + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the assertion (on the equality) can be inverted (the + implicit assertion on the existence of the path is not affected). + """ assert_attribute_matches(output, path, attribute, re.escape(text) + "$", negate=negate) def assert_element_text( - output: str, path: str, verify_assertions_function, children, negate: Union[bool, str] = False + output: Output, + path: Path, + verify_assertions_function: VerifyAssertionsFunction, + children: ChildAssertions, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """Recursively checks the specified assertions against the text of - the first element matching the specified path.""" + """This tag allows the developer to recurisively specify additional assertions as + child elements about just the text contained in the element specified by the + XPath-like ``path``, e.g. + + ```xml + + + + ``` + + The assertion implicitly also asserts that an element matching ``path`` exists. + With ``negate`` the result of the implicit assertions can be inverted. + The sub-assertions, which have their own ``negate`` attribute, are not affected + by ``negate``. + """ assert_xml_element(output, path, verify_assertions_function, children, negate=negate) def assert_xml_element( - output: str, - path: str, - verify_assertions_function=None, - children=None, - attribute: Optional[str] = None, - all: Union[bool, str] = False, - n: Optional[Union[int, str]] = None, - delta: Union[int, str] = 0, - min: Optional[Union[int, str]] = None, - max: Optional[Union[int, str]] = None, - negate: Union[bool, str] = False, + output: Output, + path: Path, + verify_assertions_function: Optional[VerifyAssertionsFunction] = None, + children: ChildAssertions = None, + attribute: OptionalAttribute = None, + all: All = False, + n: N = None, + delta: Delta = 0, + min: Min = None, + max: Max = None, + negate: Negate = NEGATE_DEFAULT, ) -> None: - """ - Check if path occurs in the xml. If n and delta or min and max are given - also the number of occurences is checked. - If there are any sub assertions then check them against - - the element's text if attribute is None - - the content of the attribute - If all is True then the sub assertions are checked for all occurences. + """Assert if the XML file contains element(s) or tag(s) with the specified + [XPath-like ``path``](https://lxml.de/xpathxslt.html). If ``n`` and ``delta`` + or ``min`` and ``max`` are given also the number of occurences is checked. + + ```xml + + + + + + ``` + + With ``negate="true"`` the outcome of the assertions wrt the precence and number + of ``path`` can be negated. If there are any sub assertions then check them against + + - the content of the attribute ``attribute`` + - the element's text if no attribute is given + + ```xml + + + + + + ``` + + Sub-assertions are not subject to the ``negate`` attribute of ``xml_element``. + If ``all`` is ``true`` then the sub assertions are checked for all occurences. + + Note that all other XML assertions can be expressed by this assertion (Galaxy + also implements the other assertions by calling this one). """ children = children or [] all = asbool(all) @@ -131,7 +280,9 @@ def assert_xml_element( else: content = occ.attrib[attribute] # type: ignore[assignment] # https://github.com/lxml/lxml-stubs/pull/99 try: - verify_assertions_function(content, children) + if content is None: + raise AssertionError("Failed to find expected XML content") + verify_assertions_function(content.encode("utf-8"), children) except AssertionError as e: if attribute is not None and attribute != "": raise AssertionError(f"Attribute '{attribute}' on element with path '{path}': {str(e)}") diff --git a/lib/galaxy/tool_util/verify/codegen.py b/lib/galaxy/tool_util/verify/codegen.py new file mode 100644 index 000000000000..7219d40b6dcb --- /dev/null +++ b/lib/galaxy/tool_util/verify/codegen.py @@ -0,0 +1,420 @@ +#!/usr/bin/env python + +# how to use this function... +# PYTHONPATH=lib python lib/galaxy/tool_util/verify/codegen.py + +from __future__ import annotations + +import argparse +import inspect +import os +from shutil import move +from typing import ( + cast, + List, + Optional, + Union, +) + +import lxml.etree as ET +from jinja2 import Environment +from typing_extensions import ( + Annotated, + get_args, + get_origin, + Literal, +) + +from galaxy.tool_util.verify.asserts import assertion_module_and_functions +from galaxy.tool_util.verify.asserts._types import AssertionParameter as AssertionParameterAnnotation +from galaxy.util.commands import shell + +models_path = os.path.join(os.path.dirname(__file__), "assertion_models.py") +galaxy_xsd_path = os.path.join(os.path.dirname(__file__), "..", "xsd", "galaxy.xsd") + +Children = Literal["allowed", "required", "forbidden"] + +DESCRIPTION = """This script synchronizes dynamic code aritfacts against models in Galaxy. + +Right now this just synchronizes Galaxy's XSD file against documentation in Galaxy's assertion modules but +in the future it will also build Pydantic models for these functions. +""" + +assert_models_template = """ +'''This module is auto-generated, please do not modify.''' +import typing +import re + +from typing_extensions import ( + Annotated, + Literal, +) + +from pydantic import ( + BaseModel, + BeforeValidator, + ConfigDict, + Field, + RootModel, + StrictFloat, + StrictInt, +) + +BYTES_PATTERN = re.compile(r"^(0|[1-9][0-9]*)([kKMGTPE]i?)?$") + + +class AssertionModel(BaseModel): + model_config = ConfigDict( + extra="forbid", + ) + + +def check_bytes(v: typing.Any) -> typing.Any: + if isinstance(v, str): + assert BYTES_PATTERN.match(v), "Not valid bytes string" + return v + + +def check_center_of_mass(v: typing.Any): + assert isinstance(v, str) + split_parts = v.split(",") + assert len(split_parts) == 2 + for part in split_parts: + assert float(part.strip()) + return v + + +def check_regex(v: typing.Any): + assert isinstance(v, str) + try: + re.compile(typing.cast(str, v)) + except re.error: + raise AssertionError(f"Invalid regular expression {v}") + return v + + +def check_non_negative_if_set(v: typing.Any): + if v is not None: + try: + assert v >= 0 + except TypeError: + raise AssertionError(f"Invalid type found {v}") + return v + + +def check_non_negative_if_int(v: typing.Any): + if v is not None and isinstance(v, int): + assert typing.cast(int, v) >= 0 + return v + + +{% for assertion in assertions %} +{% for parameter in assertion.parameters %} +{{assertion.name}}_{{ parameter.name }}_description = '''{{ parameter.description }}''' +{% endfor %} + +class {{assertion.name}}_model(AssertionModel): + r\"\"\"{{ assertion.docstring }}\"\"\" + that: Literal["{{assertion.name}}"] = "{{assertion.name}}" +{% for parameter in assertion.parameters %} +{% if not parameter.is_deprecated %} + {{ parameter.name }}: {{ parameter.type_str }} = Field( + {{ parameter.field_default_str }}, + description={{ assertion.name }}_{{ parameter.name }}_description, + ) +{% endif %} +{% endfor %} +{% if assertion.children == "required" %} + children: "assertion_list" +{% endif %} +{% if assertion.children == "allowed" %} + children: typing.Optional["assertion_list"] = None +{% endif %} +{% endfor %} + +any_assertion_model = Annotated[typing.Union[ +{% for assertion in assertions %} + {{assertion.name}}_model, +{% endfor %} +], Field(discriminator="that")] + +assertion_list = RootModel[typing.List[any_assertion_model]] +""" + + +def get_default_args(func): + signature = inspect.signature(func) + return {k: v.default for k, v in signature.parameters.items()} + + +def main(): + assertions = [] + for function_name, (module_and_function, assertion_function) in assertion_module_and_functions.items(): + if not function_name.startswith("assert_"): + continue + name = function_name[len("assert_") :] + docstring = inspect.cleandoc(assertion_function.__doc__ or "") + annotations = assertion_function.__annotations__ + default_args = get_default_args(assertion_function) + parameters = [] + children: Children = "forbidden" + for parameter_name, parameter_type in annotations.items(): + if parameter_name == "return": + continue + elif parameter_name == "children": + if default_args.get("children", inspect._empty) is not inspect._empty: + children = "allowed" + else: + children = "required" + continue + elif parameter_name in ["output", "output_bytes", "verify_assertions_function"]: + continue + + default_value = default_args.get(parameter_name) + parameters.append(AssertionParameter(parameter_name, parameter_type, default_value)) + assertion = Assertion(name, docstring, parameters, children, module_and_function) + assertions.append(assertion) + rewrite_galaxy_xsd(assertions) + write_assertion_models(assertions) + + +def _expand_template(template_str: str, assertions) -> str: + template = Environment().from_string(template_str) + return template.render(assertions=assertions) + + +def write_assertion_models(assertions): + models_file_contents = _expand_template(assert_models_template, assertions) + with open(models_path, "w") as f: + f.write(models_file_contents) + shell(["isort", models_path]) + shell(["black", models_path]) + + +def get_annotation(text: str, nsmap): + annotation = ET.Element("{http://www.w3.org/2001/XMLSchema}annotation") + documentation = ET.Element("{http://www.w3.org/2001/XMLSchema}documentation") + documentation.text = ET.CDATA(text) + documentation.attrib["{http://www.w3.org/XML/1998/namespace}lang"] = "en" + annotation.append(documentation) + return annotation + + +def parameter_xsd_element(assertion_parameter, nsmap): + el = ET.Element("{http://www.w3.org/2001/XMLSchema}attribute", nsmap=nsmap) + el.attrib["name"] = assertion_parameter.name + el.attrib["type"] = assertion_parameter.xml_type_str + el.attrib["use"] = "required" if not assertion_parameter.has_default_value else "optional" + annotation = get_annotation(assertion_parameter.description, nsmap) + el.append(annotation) + return el + + +def xsd_element(assertion, nsmap): + el = ET.Element("{http://www.w3.org/2001/XMLSchema}element", nsmap=nsmap) + el.attrib["name"] = assertion.name + annotation = get_annotation(assertion.docstring + "\n\n$attribute_list::5", nsmap) + complexType = ET.Element("{http://www.w3.org/2001/XMLSchema}complexType", nsmap=nsmap) + if assertion.children != "forbidden": + sequence = ET.Element("{http://www.w3.org/2001/XMLSchema}sequence", nsmap=nsmap) + group = ET.Element("{http://www.w3.org/2001/XMLSchema}group", nsmap=nsmap) + group.attrib["ref"] = "TestAssertion" + group.attrib["minOccurs"] = "0" if assertion.children == "allowed" else "1" + group.attrib["maxOccurs"] = "unbounded" + sequence.append(group) + complexType.append(sequence) + for parameter in assertion.parameters: + complexType.append(parameter_xsd_element(parameter, nsmap)) + el.append(annotation) + el.append(complexType) + return el + + +def xsd_elements(assertions, nsmap): + elements = [] + for assertion in assertions: + elements.append(ET.Comment(f" XSD doc for element auto-generated from {assertion.module_and_function}")) + elements.append(xsd_element(assertion, nsmap)) + return elements + + +def rewrite_galaxy_xsd(assertions): + with open(galaxy_xsd_path, "rb") as f: + contents = f.read() + parser = ET.XMLParser(strip_cdata=False) + root = ET.fromstring(contents, parser=parser) + raw_element = root.find( + ".//{http://www.w3.org/2001/XMLSchema}group[@name='TestAssertion']/{http://www.w3.org/2001/XMLSchema}choice" + ) + assert raw_element + element = cast(ET._Element, raw_element) + for el in element.iterchildren(): + element.remove(el) + element.append(ET.Comment("The following block and all children are auto-generated - please do not modify.")) + for xsd_element in xsd_elements(assertions, root.nsmap): + xsd_element.tail = "\n " + element.append(xsd_element) + xml_new = ET.tostring(root).decode("utf-8") + with open(galaxy_xsd_path, "w") as f: + f.write('\n') + f.write(xml_new) + f.write("\n") + ret_code = shell(["xmllint", "--format", "--output", "galaxy-tmp.xsd", galaxy_xsd_path]) + assert ret_code == 0 + move("galaxy-tmp.xsd", galaxy_xsd_path) + + +class AssertionParameter: + + def __init__(self, name: str, type: str, default_value): + self.name = name + self.type = type + self.default_value = default_value + + @property + def description(self) -> str: + type = self.type + if hasattr(type, "__metadata__"): + return type.__metadata__[0].doc + else: + return "" + + @property + def type_str(self) -> str: + raw_type_str = as_type_str(self.type) + validators = self.validators[:] + if self.xml_type_str == "Bytes": + validators.append("check_bytes") + validators.append("check_non_negative_if_int") + if len(validators) > 0: + validation_str = ",".join([f"BeforeValidator({v})" for v in validators]) + return f"Annotated[{raw_type_str}, {validation_str}]" + + return raw_type_str + + @property + def xml_type_str(self) -> str: + return as_xml_type(self.type) + + @property + def field_default_str(self) -> str: + if not self.has_default_value: + return "..." + elif isinstance(self.default_value, str): + return f'''"{self.default_value}"''' + else: + return str(self.default_value) + + @property + def has_default_value(self) -> bool: + return self.default_value is not inspect._empty + + @property + def is_deprecated(self) -> bool: + assertion_parameter = self._get_type_annotation() + if assertion_parameter is not None: + return assertion_parameter.deprecated + + return False + + @property + def validators(self) -> List[str]: + assertion_parameter = self._get_type_annotation() + if assertion_parameter is not None: + return assertion_parameter.validators + + return [] + + def _get_type_annotation(self) -> Optional[AssertionParameterAnnotation]: + target_type = self.type + if get_origin(target_type) is Annotated: + args = get_args(target_type) + if len(args) > 1: + return cast(AssertionParameterAnnotation, args[1]) + + return None + + +def _is_none_type(target_type): + return target_type is type(None) + + +def _non_optional_types(union_type): + return [t for t in get_args(union_type) if not _is_none_type(t)] + + +def as_xml_type(target_type) -> str: + if get_origin(target_type) is Annotated: + args = get_args(target_type) + if len(args) > 1: + assertion_parameter = args[1] + if assertion_parameter.xml_type: + return assertion_parameter.xml_type + + return as_xml_type(args[0]) + elif get_origin(target_type) is Union: + types = _non_optional_types(target_type) + if len(types) == 2: + non_str_types = [t for t in types if t is not str] + if len(non_str_types) == 1 and non_str_types[0] is bool: + return "xs:boolean" + elif len(non_str_types) == 1 and non_str_types[0] is int: + return "xs:integer" + elif len(non_str_types) == 1 and non_str_types[0] is float: + return "xs:float" + + return "xs:string" + + +def as_type_str(target_type): + if get_origin(target_type) is Annotated: + args = get_args(target_type) + if len(args) > 1: + if args[1].json_type: + return args[1].json_type + + return as_type_str(args[0]) + elif get_origin(target_type) is Union: + is_optional = any(_is_none_type(t) for t in get_args(target_type)) + types_as_str = ", ".join(map(as_type_str, _non_optional_types(target_type))) + union_type = f"typing.Union[{types_as_str}]" + if is_optional: + return f"typing.Optional[{union_type}]" + else: + return union_type + elif target_type is str: + return "str" + elif target_type is int: + return "int" + elif target_type is float: + return "float" + elif target_type is bool: + return "bool" + else: + return str(target_type) + + +class Assertion: + + def __init__( + self, + name: str, + docstring: str, + parameters: List[AssertionParameter], + children: Children, + module_and_function: str, + ): + self.name = name + self.parameters = parameters + self.docstring = docstring + self.children = children + self.module_and_function = module_and_function + + +def arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=DESCRIPTION) + return parser + + +if __name__ == "__main__": + main() diff --git a/lib/galaxy/tool_util/xsd/galaxy.xsd b/lib/galaxy/tool_util/xsd/galaxy.xsd index ec549456dbc5..b8d61030bb05 100644 --- a/lib/galaxy/tool_util/xsd/galaxy.xsd +++ b/lib/galaxy/tool_util/xsd/galaxy.xsd @@ -1,15 +1,9 @@ - + Galaxy Schema A Galaxy XML tool wrapper - - - + + - + - + @@ -136,7 +130,7 @@ the tool menu immediately following the hyperlink for the tool (based on the - + @@ -244,7 +238,6 @@ this attribute defined the HTTP request method to use when communicating with an - Frequently, tools may require the same XML @@ -266,8 +259,7 @@ Example: macros.xml -````]]> - +````]]> @@ -287,14 +279,12 @@ Usage example: mypackage -````]]> - +````]]> - - - - ` is identical to ``, `` is identical to `