mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Merge branch 'release_22.05' into release_23.0
This commit is contained in:
@@ -185,23 +185,11 @@ def _check_asserts(test_idx, assertions, lint_ctx):
|
||||
lint_ctx.error(f"Test {test_idx}: unknown assertion '{a.tag}'", node=a)
|
||||
continue
|
||||
assert_function_sig = signature(asserts.assertion_functions[assert_function_name])
|
||||
# check type of the attributes (int, float ...)
|
||||
# check of the attributes
|
||||
for attrib in a.attrib:
|
||||
if attrib not in assert_function_sig.parameters:
|
||||
lint_ctx.error(f"Test {test_idx}: unknown attribute '{attrib}' for '{a.tag}'", node=a)
|
||||
continue
|
||||
annotation = assert_function_sig.parameters[attrib].annotation
|
||||
annotation = _handle_optionals(annotation)
|
||||
if annotation is not Parameter.empty:
|
||||
try:
|
||||
annotation(a.attrib[attrib])
|
||||
except TypeError:
|
||||
raise Exception(f"Faild to instantiate {attrib} for {assert_function_name}")
|
||||
except ValueError:
|
||||
lint_ctx.error(
|
||||
f"Test {test_idx}: attribute '{attrib}' for '{a.tag}' needs to be '{annotation.__name__}' got '{a.attrib[attrib]}'",
|
||||
node=a,
|
||||
)
|
||||
# check missing required attributes
|
||||
for p in assert_function_sig.parameters:
|
||||
if p in ["output", "output_bytes", "verify_assertions_function", "children"]:
|
||||
|
||||
@@ -60,7 +60,7 @@ def verify_assertion(data: bytes, assertion_description):
|
||||
# output. children is the parsed version of the child elements of
|
||||
# the XML element describing this assertion. See
|
||||
# assert_element_text in test/base/asserts/xml.py as an example of
|
||||
# how to use verify_assertions_function and children in conjuction
|
||||
# how to use verify_assertions_function and children in conjunction
|
||||
# to apply assertion checking to a subset of the input. The parsed
|
||||
# version of an elements child elements do not need to just define
|
||||
# assertions, developers of assertion functions can also use the
|
||||
|
||||
@@ -1,10 +1,25 @@
|
||||
from math import inf
|
||||
from typing import (
|
||||
Callable,
|
||||
Optional,
|
||||
TypeVar,
|
||||
Union,
|
||||
)
|
||||
|
||||
from galaxy.util import asbool
|
||||
from galaxy.util.bytesize import parse_bytesize
|
||||
|
||||
|
||||
def _assert_number(count, n, delta, min, max, negate, n_text, min_max_text):
|
||||
def _assert_number(
|
||||
count: int,
|
||||
n: Optional[Union[int, str]],
|
||||
delta: Union[int, str],
|
||||
min: Optional[Union[int, str]],
|
||||
max: Optional[Union[int, str]],
|
||||
negate: Union[bool, str],
|
||||
n_text: str,
|
||||
min_max_text: str,
|
||||
) -> None:
|
||||
"""
|
||||
helper function for assering that count is in
|
||||
- [n-delta:n+delta]
|
||||
@@ -26,12 +41,12 @@ def _assert_number(count, n, delta, min, max, negate, n_text, min_max_text):
|
||||
)
|
||||
if min is not None or max is not None:
|
||||
if min is None:
|
||||
min = -inf # also replacing min/max for output
|
||||
min = "-inf" # also replacing min/max for output
|
||||
min_bytes = -inf
|
||||
else:
|
||||
min_bytes = parse_bytesize(min)
|
||||
if max is None:
|
||||
max = inf
|
||||
max = "inf"
|
||||
max_bytes = inf
|
||||
else:
|
||||
max_bytes = parse_bytesize(max)
|
||||
@@ -41,9 +56,24 @@ def _assert_number(count, n, delta, min, max, negate, n_text, min_max_text):
|
||||
)
|
||||
|
||||
|
||||
OutputType = TypeVar("OutputType")
|
||||
TextType = TypeVar("TextType")
|
||||
|
||||
|
||||
def _assert_presence_number(
|
||||
output, text, n, delta, min, max, negate, check_presence_foo, count_foo, presence_text, n_text, min_max_text
|
||||
):
|
||||
output: OutputType,
|
||||
text: TextType,
|
||||
n: Optional[Union[int, str]],
|
||||
delta: Union[int, str],
|
||||
min: Optional[Union[int, str]],
|
||||
max: Optional[Union[int, str]],
|
||||
negate: Union[bool, str],
|
||||
check_presence_foo: Callable[[OutputType, TextType], bool],
|
||||
count_foo: Callable[[OutputType, TextType], int],
|
||||
presence_text: str,
|
||||
n_text: str,
|
||||
min_max_text: str,
|
||||
) -> None:
|
||||
"""
|
||||
helper function to assert that
|
||||
- text is present in output using check_presence_foo
|
||||
|
||||
@@ -3,14 +3,17 @@ import re
|
||||
import tarfile
|
||||
import tempfile
|
||||
import zipfile
|
||||
from typing import Optional
|
||||
from typing import (
|
||||
Optional,
|
||||
Union,
|
||||
)
|
||||
|
||||
from galaxy.util import asbool
|
||||
from ._util import _assert_presence_number
|
||||
|
||||
|
||||
def _extract_from_tar(bytes, fn):
|
||||
with io.BytesIO(bytes) as temp:
|
||||
def _extract_from_tar(output_bytes, fn):
|
||||
with io.BytesIO(output_bytes) as temp:
|
||||
with tarfile.open(fileobj=temp, mode="r") as tar_temp:
|
||||
ti = tar_temp.getmember(fn)
|
||||
# zip treats directories like empty files.
|
||||
@@ -23,9 +26,9 @@ def _extract_from_tar(bytes, fn):
|
||||
return member_fh.read()
|
||||
|
||||
|
||||
def _list_from_tar(bytes, path):
|
||||
def _list_from_tar(output_bytes, path):
|
||||
lst = list()
|
||||
with io.BytesIO(bytes) as temp:
|
||||
with io.BytesIO(output_bytes) as temp:
|
||||
with tarfile.open(fileobj=temp, mode="r") as tar_temp:
|
||||
for fn in tar_temp.getnames():
|
||||
if not re.match(path, fn):
|
||||
@@ -34,16 +37,16 @@ def _list_from_tar(bytes, path):
|
||||
return sorted(lst)
|
||||
|
||||
|
||||
def _extract_from_zip(bytes, fn):
|
||||
with io.BytesIO(bytes) as temp:
|
||||
def _extract_from_zip(output_bytes, fn):
|
||||
with io.BytesIO(output_bytes) as temp:
|
||||
with zipfile.ZipFile(temp, mode="r") as zip_temp:
|
||||
with zip_temp.open(fn) as member_fh:
|
||||
return member_fh.read()
|
||||
|
||||
|
||||
def _list_from_zip(bytes, path):
|
||||
def _list_from_zip(output_bytes, path):
|
||||
lst = list()
|
||||
with io.BytesIO(bytes) as temp:
|
||||
with io.BytesIO(output_bytes) as temp:
|
||||
with zipfile.ZipFile(temp, mode="r") as zip_temp:
|
||||
for fn in zip_temp.namelist():
|
||||
if not re.match(path, fn):
|
||||
@@ -53,17 +56,17 @@ def _list_from_zip(bytes, path):
|
||||
|
||||
|
||||
def assert_has_archive_member(
|
||||
output_bytes,
|
||||
path,
|
||||
output_bytes: bytes,
|
||||
path: str,
|
||||
verify_assertions_function,
|
||||
children,
|
||||
all="false",
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> 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."""
|
||||
|
||||
@@ -13,7 +13,7 @@ def _assert_h5py():
|
||||
raise Exception(IMPORT_MISSING_MESSAGE)
|
||||
|
||||
|
||||
def assert_has_h5_attribute(output_bytes, key, value):
|
||||
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"""
|
||||
_assert_h5py()
|
||||
@@ -25,11 +25,10 @@ def assert_has_h5_attribute(output_bytes, key, value):
|
||||
|
||||
|
||||
# TODO the function actually queries groups. so the function and argument name are misleading
|
||||
def assert_has_h5_keys(output_bytes, keys):
|
||||
def assert_has_h5_keys(output_bytes: bytes, keys: str) -> None:
|
||||
"""Asserts the specified HDF5 output has the given keys."""
|
||||
_assert_h5py()
|
||||
keys = [k.strip() for k in keys.strip().split(",")]
|
||||
h5_keys = sorted(keys)
|
||||
h5_keys = sorted([k.strip() for k in keys.strip().split(",")])
|
||||
output_temp = io.BytesIO(output_bytes)
|
||||
local_keys = []
|
||||
|
||||
|
||||
@@ -1,16 +1,19 @@
|
||||
from typing import Optional
|
||||
from typing import (
|
||||
Optional,
|
||||
Union,
|
||||
)
|
||||
|
||||
from ._util import _assert_number
|
||||
|
||||
|
||||
def assert_has_size(
|
||||
output_bytes,
|
||||
value: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
output_bytes: bytes,
|
||||
value: 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,
|
||||
) -> None:
|
||||
"""
|
||||
Asserts the specified output has a size of the specified value,
|
||||
allowing for absolute (delta) and relative (delta_frac) difference.
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
import re
|
||||
from typing import Optional
|
||||
from typing import (
|
||||
Optional,
|
||||
Union,
|
||||
)
|
||||
|
||||
from ._util import _assert_number
|
||||
|
||||
|
||||
def get_first_line(output, comment):
|
||||
def get_first_line(output: str, comment: str) -> str:
|
||||
"""
|
||||
get the first non-comment and non-empty line
|
||||
"""
|
||||
@@ -19,15 +22,15 @@ def get_first_line(output, comment):
|
||||
|
||||
|
||||
def assert_has_n_columns(
|
||||
output,
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
sep="\t",
|
||||
comment="",
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> 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
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
import re
|
||||
from typing import Optional
|
||||
from typing import (
|
||||
Optional,
|
||||
Union,
|
||||
)
|
||||
|
||||
from ._util import (
|
||||
_assert_number,
|
||||
@@ -8,14 +11,14 @@ from ._util import (
|
||||
|
||||
|
||||
def assert_has_text(
|
||||
output,
|
||||
text,
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> None:
|
||||
"""Asserts specified output contains the substring specified by
|
||||
the argument text. The exact number of occurrences can be
|
||||
optionally specified by the argument n"""
|
||||
@@ -36,7 +39,7 @@ def assert_has_text(
|
||||
)
|
||||
|
||||
|
||||
def assert_not_has_text(output, text):
|
||||
def assert_not_has_text(output: str, text: str) -> 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)"
|
||||
@@ -44,14 +47,14 @@ def assert_not_has_text(output, text):
|
||||
|
||||
|
||||
def assert_has_line(
|
||||
output,
|
||||
line,
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> None:
|
||||
"""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"""
|
||||
@@ -73,13 +76,13 @@ def assert_has_line(
|
||||
|
||||
|
||||
def assert_has_n_lines(
|
||||
output,
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> None:
|
||||
"""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"""
|
||||
@@ -98,14 +101,14 @@ def assert_has_n_lines(
|
||||
|
||||
|
||||
def assert_has_text_matching(
|
||||
output,
|
||||
expression,
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> None:
|
||||
"""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)
|
||||
@@ -128,14 +131,14 @@ def assert_has_text_matching(
|
||||
|
||||
|
||||
def assert_has_line_matching(
|
||||
output,
|
||||
expression,
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> None:
|
||||
"""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."""
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
import re
|
||||
from typing import Optional
|
||||
from typing import (
|
||||
Optional,
|
||||
Union,
|
||||
)
|
||||
|
||||
from lxml.etree import XMLSyntaxError
|
||||
|
||||
@@ -11,7 +14,7 @@ from galaxy.util import (
|
||||
)
|
||||
|
||||
|
||||
def assert_is_valid_xml(output):
|
||||
def assert_is_valid_xml(output: str) -> None:
|
||||
"""Simple assertion that just verifies the specified output
|
||||
is valid XML."""
|
||||
try:
|
||||
@@ -20,7 +23,7 @@ def assert_is_valid_xml(output):
|
||||
raise AssertionError(f"Expected valid XML, but could not parse output. {unicodify(e)}")
|
||||
|
||||
|
||||
def assert_has_element_with_path(output, path, negate: bool = False):
|
||||
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;
|
||||
@@ -29,64 +32,68 @@ def assert_has_element_with_path(output, path, negate: bool = False):
|
||||
|
||||
|
||||
def assert_has_n_elements_with_path(
|
||||
output,
|
||||
path,
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> None:
|
||||
"""Asserts the specified output has exactly n elements matching the
|
||||
path specified."""
|
||||
assert_xml_element(output, path, n=n, delta=delta, min=min, max=max, negate=negate)
|
||||
|
||||
|
||||
def assert_element_text_matches(output, path, expression, negate: bool = False):
|
||||
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."""
|
||||
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, path, text, negate: bool = False):
|
||||
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."""
|
||||
assert_element_text_matches(output, path, re.escape(text) + "$", negate=negate)
|
||||
|
||||
|
||||
def assert_attribute_matches(output, path, attribute, expression, negate: bool = False):
|
||||
def assert_attribute_matches(
|
||||
output: str, path: str, attribute, expression: str, negate: Union[bool, str] = False
|
||||
) -> None:
|
||||
"""Asserts the specified attribute of the first element matching
|
||||
the specified path matches the specified regular expression."""
|
||||
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, path, attribute, text, negate: bool = False):
|
||||
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."""
|
||||
assert_attribute_matches(output, path, attribute, re.escape(text) + "$", negate=negate)
|
||||
|
||||
|
||||
def assert_element_text(output, path, verify_assertions_function, children, negate: bool = False):
|
||||
def assert_element_text(
|
||||
output: str, path: str, verify_assertions_function, children, negate: Union[bool, str] = False
|
||||
) -> None:
|
||||
"""Recursively checks the specified assertions against the text of
|
||||
the first element matching the specified path."""
|
||||
assert_xml_element(output, path, verify_assertions_function, children, negate=negate)
|
||||
|
||||
|
||||
def assert_xml_element(
|
||||
output,
|
||||
path,
|
||||
output: str,
|
||||
path: str,
|
||||
verify_assertions_function=None,
|
||||
children=None,
|
||||
attribute=None,
|
||||
all=False,
|
||||
n: Optional[int] = None,
|
||||
delta: int = 0,
|
||||
min: Optional[int] = None,
|
||||
max: Optional[int] = None,
|
||||
negate: bool = False,
|
||||
):
|
||||
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,
|
||||
) -> 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.
|
||||
|
||||
@@ -1644,8 +1644,7 @@ def test_tests_asserts(lint_ctx):
|
||||
assert "Test 1: unknown assertion 'invalid'" in lint_ctx.error_messages
|
||||
assert "Test 1: unknown attribute 'invalid_attrib' for 'has_text'" in lint_ctx.error_messages
|
||||
assert "Test 1: missing attribute 'text' for 'has_text'" in lint_ctx.error_messages
|
||||
assert "Test 1: attribute 'value' for 'has_size' needs to be 'int' got '500k'" in lint_ctx.error_messages
|
||||
assert "Test 1: attribute 'delta' for 'has_size' needs to be 'int' got '1O'" in lint_ctx.error_messages
|
||||
assert "Test 1: attribute 'value' for 'has_size' needs to be 'int' got '500k'" not in lint_ctx.error_messages
|
||||
assert (
|
||||
"Test 1: unknown attribute 'invalid_attrib_also_checked_in_nested_asserts' for 'not_has_text'"
|
||||
in lint_ctx.error_messages
|
||||
@@ -1654,7 +1653,7 @@ def test_tests_asserts(lint_ctx):
|
||||
assert "Test 1: 'has_n_columns' needs to specify 'n', 'min', or 'max'" in lint_ctx.error_messages
|
||||
assert "Test 1: 'has_n_lines' needs to specify 'n', 'min', or 'max'" in lint_ctx.error_messages
|
||||
assert not lint_ctx.warn_messages
|
||||
assert len(lint_ctx.error_messages) == 9
|
||||
assert len(lint_ctx.error_messages) == 7
|
||||
|
||||
|
||||
def test_tests_output_type_mismatch(lint_ctx):
|
||||
|
||||
Reference in New Issue
Block a user