Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.tool_util.verify.asserts.archive
import io
import re
import tarfile
import tempfile
import zipfile
from typing import (
Optional,
Union,
)
from galaxy.util import asbool
from ._util import _assert_presence_number
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.
# so make this consistent for tar
if ti.isdir():
return ""
tar_file = tar_temp.extractfile(fn)
assert tar_file is not None
with tar_file as member_fh:
return member_fh.read()
def _list_from_tar(output_bytes, path):
lst = list()
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):
continue
lst.append(fn)
return sorted(lst)
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(output_bytes, path):
lst = list()
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):
continue
lst.append(fn)
return sorted(lst)
[docs]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,
) -> 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."""
all = asbool(all)
extract_foo = None
# from python 3.9 is_tarfile supports file like objects then we do not need
# the tempfile detour but can use io.BytesIO(output_bytes)
with tempfile.NamedTemporaryFile() as tmp:
tmp.write(output_bytes)
tmp.flush()
if zipfile.is_zipfile(tmp.name):
extract_foo = _extract_from_zip
list_foo = _list_from_zip
elif tarfile.is_tarfile(tmp.name):
extract_foo = _extract_from_tar
list_foo = _list_from_tar
assert extract_foo is not None, f"Expected path '{path}' to be an archive"
# get list of matching file names in archive and check against n, delta,
# min, max (slightly abusing the output and text as well as the function
# parameters)
fns = list_foo(output_bytes, path)
_assert_presence_number(
None,
path,
n,
delta,
min,
max,
negate,
lambda o, t: len(fns) > 0,
lambda o, t: len(fns),
"{expected} path '{text}' in archive",
"{expected} {n}+-{delta} matches for path '{text}' in archive",
"{expected} that the number of matches for path '{text}' in archive is in [{min}:{max}]",
)
# check sub-assertions on members matching path
for fn in fns:
contents = extract_foo(output_bytes, fn)
try:
verify_assertions_function(contents, children)
except AssertionError as e:
raise AssertionError(f"Archive member '{path}': {str(e)}")
if not all:
break