Merge pull request #15873 from bernt-matthias/topic/fix-assertion-bytes (012102dc) · Commits · NDIP / Galaxy

lib/galaxy/tool_util/linters/tests.py

+1 −13

Original line number	Diff line number	Diff line
		@@ -152,23 +152,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"]:

lib/galaxy/tool_util/verify/asserts/init.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -60,7 +60,7 @@ def verify_assertion(data, 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

lib/galaxy/tool_util/verify/asserts/_util.py

+35 −5

Original line number	Diff line number	Diff line
		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

lib/galaxy/tool_util/verify/asserts/archive.py

+21 −18

Original line number	Diff line number	Diff line
		@@ -3,14 +3,17 @@ import re
		import tarfile
		import tempfile
		import zipfile
		from typing import Optional
		from typing import (
		Union,
		Optional,
		)

		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.
		@@ -21,9 +24,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):
		@@ -32,16 +35,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):
		@@ -51,17 +54,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."""

lib/galaxy/tool_util/verify/asserts/hdf5.py

+3 −4

Original line number	Diff line number	Diff line
		@@ -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 = []