Source code for planemo.test._check_output

"""Check an output file from a generalize artifact test."""

import os
import tempfile

import requests
from galaxy.tool_util.parser.interface import TestCollectionOutputDef
from galaxy.tool_util.verify import verify
from galaxy.tool_util.verify.interactor import verify_collection
from galaxy.util import unicodify


[docs]def check_output(runnable, output_properties, test_properties, **kwds): """Use galaxy-tool-util to check a test output. Return a list of strings describing the problems encountered, and empty list indicates no problems were detected. Currently this will only ever return at most one detected problem because of the way galaxy-tool-util throws exceptions instead of returning individual descriptions - but this may be enhanced in the future. """ checker = _check_output_collection if for_collections(test_properties) else _check_output_file return checker(runnable, output_properties, test_properties, **kwds)
[docs]def for_collections(test_properties): return "element_tests" in test_properties
def _check_output_collection(runnable, output_properties, test_properties, **kwds): data_collection = output_properties output_def = TestCollectionOutputDef.from_dict(test_properties) def verify_dataset(element, element_attrib, element_outfile): if element_outfile: element_attrib["path"] = element_outfile _verify_output_file(runnable, element["_output_object"], element_attrib) problems = [] try: verify_collection(output_def, data_collection, verify_dataset) except AssertionError as e: problems.append(unicodify(e)) return problems def _verify_output_file(runnable, output_properties, test_properties, **kwds): get_filename = _test_filename_getter(runnable) path = output_properties["path"] with open(path, "rb") as fh: output_content = fh.read() # Support Galaxy-like file location (using "file") or CWL-like ("path" or "location"). expected_file = test_properties.get("file", None) if expected_file is None: expected_file = test_properties.get("path", None) if expected_file is None: location = test_properties.get("location") if location: if location.startswith(("http://", "https://")): expected_file = _get_location(location) else: expected_file = location.split("file://", 1)[-1] job_output_files = kwds.get("job_output_files", None) item_label = "Output with path %s" % path if "asserts" in test_properties: # TODO: break fewer abstractions here... from galaxy.tool_util.parser.yaml import __to_test_assert_list test_properties["assert_list"] = __to_test_assert_list(test_properties["asserts"]) verify( item_label, output_content, attributes=test_properties, filename=expected_file, get_filename=get_filename, keep_outputs_dir=job_output_files, verify_extra_files=None, ) def _check_output_file(runnable, output_properties, test_properties, **kwds): problems = [] try: _verify_output_file(runnable, output_properties, test_properties, **kwds) except AssertionError as e: problems.append(unicodify(e)) return problems def _get_location(location): data_file = tempfile.NamedTemporaryFile(prefix="planemo_test_file_", delete=False) with requests.get(location, stream=True) as r: r.raise_for_status() for chunk in r.iter_content(): if chunk: data_file.write(chunk) return data_file.name def _test_filename_getter(runnable): def get_filename(name): artifact_directory = os.path.dirname(runnable.path) return os.path.join(artifact_directory, name) return get_filename __all__ = ("check_output",)