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.script
#!/usr/bin/env python
import argparse
import datetime as dt
import json
import logging
import os
import sys
import tempfile
from collections import namedtuple
from concurrent.futures import (
thread,
ThreadPoolExecutor,
)
import yaml
from galaxy.tool_util.verify.interactor import (
DictClientTestConfig,
GalaxyInteractorApi,
verify_tool,
)
DESCRIPTION = """Script to quickly run a tool test against a running Galaxy instance."""
DEFAULT_SUITE_NAME = "Galaxy Tool Tests"
ALL_TESTS = -1
ALL_TOOLS = "*"
ALL_VERSION = "*"
LATEST_VERSION = None
TestReference = namedtuple("TestReference", ["tool_id", "tool_version", "test_index"])
TestException = namedtuple("TestException", ["tool_id", "exception", "was_recorded"])
[docs]class Results:
[docs] def __init__(self, default_suitename, test_json, append=False, galaxy_url=None):
self.test_json = test_json or "-"
self.galaxy_url = galaxy_url
test_results = []
test_exceptions = []
suitename = default_suitename
if append:
assert test_json != "-"
with open(test_json) as f:
previous_results = json.load(f)
test_results = previous_results["tests"]
if "suitename" in previous_results:
suitename = previous_results["suitename"]
self.test_results = test_results
self.test_exceptions = test_exceptions
self.suitename = suitename
[docs] def already_successful(self, test_reference):
test_data = self._previous_test_data(test_reference)
if test_data:
if "status" in test_data and test_data["status"] == "success":
return True
return False
[docs] def already_executed(self, test_reference):
test_data = self._previous_test_data(test_reference)
if test_data:
if "status" in test_data and test_data["status"] != "skipped":
return True
return False
def _previous_test_data(self, test_reference):
test_id = _test_id_for_reference(test_reference)
for test_result in self.test_results:
if test_result.get("id") != test_id:
continue
has_data = test_result.get("has_data", False)
if has_data:
test_data = test_result.get("data", {})
return test_data
return None
[docs] def write(self):
tests = sorted(self.test_results, key=lambda el: el["id"])
n_passed, n_failures, n_skips = 0, 0, 0
n_errors = len([e for e in self.test_exceptions if not e.was_recorded])
for test in tests:
has_data = test.get("has_data", False)
if has_data:
test_data = test.get("data", {})
if "status" not in test_data:
raise Exception(f"Test result data {test_data} doesn't contain a status key.")
status = test_data["status"]
if status == "success":
n_passed += 1
elif status == "error":
n_errors += 1
elif status == "skip":
n_skips += 1
elif status == "failure":
n_failures += 1
report_obj = {
"version": "0.1",
"suitename": self.suitename,
"results": {
"total": n_passed + n_failures + n_skips + n_errors,
"errors": n_errors,
"failures": n_failures,
"skips": n_skips,
},
"tests": tests,
}
if self.galaxy_url:
report_obj["galaxy_url"] = self.galaxy_url
if self.test_json == "-":
print(json.dumps(report_obj))
else:
with open(self.test_json, "w") as f:
json.dump(report_obj, f)
[docs] def info_message(self):
messages = []
passed_tests = self._tests_with_status("success")
messages.append("Passed tool tests ({}): {}".format(len(passed_tests), [t["id"] for t in passed_tests]))
failed_tests = self._tests_with_status("failure")
messages.append("Failed tool tests ({}): {}".format(len(failed_tests), [t["id"] for t in failed_tests]))
skiped_tests = self._tests_with_status("skip")
messages.append("Skipped tool tests ({}): {}".format(len(skiped_tests), [t["id"] for t in skiped_tests]))
errored_tests = self._tests_with_status("error")
messages.append("Errored tool tests ({}): {}".format(len(errored_tests), [t["id"] for t in errored_tests]))
return "\n".join(messages)
@property
def success_count(self):
self._tests_with_status("success")
@property
def skip_count(self):
self._tests_with_status("skip")
@property
def error_count(self):
return self._tests_with_status("error") + len(self.test_exceptions)
@property
def failure_count(self):
return self._tests_with_status("failure")
def _tests_with_status(self, status):
return [t for t in self.test_results if t.get("data", {}).get("status") == status]
[docs]def test_tools(
galaxy_interactor,
test_references,
results,
log=None,
parallel_tests=1,
history_per_test_case=False,
history_name=None,
no_history_reuse=False,
no_history_cleanup=False,
publish_history=False,
retries=0,
verify_kwds=None,
):
"""Run through tool tests and write report.
Refactor this into Galaxy in 21.01.
"""
verify_kwds = (verify_kwds or {}).copy()
tool_test_start = dt.datetime.now()
history_created = False
test_history = None
if not history_per_test_case:
if not history_name:
history_name = f"History for {results.suitename}"
if log:
log.info(f"History name is '{history_name}'")
if not no_history_reuse:
history = galaxy_interactor.get_history(history_name=history_name)
if history:
test_history = history["id"]
if log:
log.info(f"Using existing history with id '{test_history}', last updated: {history['update_time']}")
if not test_history:
history_created = True
test_history = galaxy_interactor.new_history(history_name=history_name, publish_history=publish_history)
if log:
log.info(f"History created with id '{test_history}'")
verify_kwds.update(
{
"no_history_cleanup": no_history_cleanup,
"test_history": test_history,
}
)
with ThreadPoolExecutor(max_workers=parallel_tests) as executor:
try:
for test_reference in test_references:
_test_tool(
executor=executor,
test_reference=test_reference,
results=results,
galaxy_interactor=galaxy_interactor,
log=log,
retries=retries,
verify_kwds=verify_kwds,
publish_history=publish_history,
)
finally:
# Always write report, even if test was cancelled.
try:
executor.shutdown(wait=True)
except KeyboardInterrupt:
executor._threads.clear()
thread._threads_queues.clear()
results.write()
if log:
if results.test_json == "-":
destination = "standard output"
else:
destination = os.path.abspath(results.test_json)
log.info(f"Report written to '{destination}'")
log.info(results.info_message())
log.info(f"Total tool test time: {dt.datetime.now() - tool_test_start}")
if history_created and not no_history_cleanup:
galaxy_interactor.delete_history(test_history)
def _test_id_for_reference(test_reference):
tool_id = test_reference.tool_id
tool_version = test_reference.tool_version
test_index = test_reference.test_index
if tool_version and tool_id.endswith(f"/{tool_version}"):
tool_id = tool_id[: -len(f"/{tool_version}")]
label_base = tool_id
if tool_version:
label_base += f"/{str(tool_version)}"
test_id = f"{label_base}-{str(test_index)}"
return test_id
def _test_tool(
executor,
test_reference,
results,
galaxy_interactor,
log,
retries,
publish_history,
verify_kwds,
):
tool_id = test_reference.tool_id
tool_version = test_reference.tool_version
test_index = test_reference.test_index
# If given a tool_id with a version suffix, strip it off so we can treat tool_version
# correctly at least in client_test_config.
if tool_version and tool_id.endswith(f"/{tool_version}"):
tool_id = tool_id[: -len(f"/{tool_version}")]
test_id = _test_id_for_reference(test_reference)
def run_test():
run_retries = retries
job_data = None
job_exception = None
def register(job_data_):
nonlocal job_data
job_data = job_data_
try:
while run_retries >= 0:
job_exception = None
try:
if log:
log.info("Executing test '%s'", test_id)
verify_tool(
tool_id,
galaxy_interactor,
test_index=test_index,
tool_version=tool_version,
register_job_data=register,
publish_history=publish_history,
**verify_kwds,
)
if log:
log.info("Test '%s' passed", test_id)
break
except Exception as e:
if log:
log.warning("Test '%s' failed", test_id, exc_info=True)
job_exception = e
run_retries -= 1
finally:
if job_data is not None:
results.register_result(
{
"id": test_id,
"has_data": True,
"data": job_data,
}
)
if job_exception is not None:
was_recorded = job_data is not None
test_exception = TestException(tool_id, job_exception, was_recorded)
results.register_exception(test_exception)
executor.submit(run_test)
[docs]def build_case_references(
galaxy_interactor,
tool_id=ALL_TOOLS,
tool_version=LATEST_VERSION,
test_index=ALL_TESTS,
page_size=0,
page_number=0,
test_filters=None,
log=None,
):
test_references = []
if tool_id == ALL_TOOLS:
tests_summary = galaxy_interactor.get_tests_summary()
for tool_id, tool_versions_dict in tests_summary.items():
for tool_version, summary in tool_versions_dict.items():
for test_index in range(summary["count"]):
test_reference = TestReference(tool_id, tool_version, test_index)
test_references.append(test_reference)
else:
assert tool_id
tool_test_dicts = galaxy_interactor.get_tool_tests(tool_id, tool_version=tool_version) or {}
for i, tool_test_dict in enumerate(tool_test_dicts):
this_tool_version = tool_test_dict.get("tool_version", tool_version)
this_test_index = i
if test_index == ALL_TESTS or i == test_index:
test_reference = TestReference(tool_id, this_tool_version, this_test_index)
test_references.append(test_reference)
if test_filters is not None and len(test_filters) > 0:
filtered_test_references = []
for test_reference in test_references:
skip_test = False
for test_filter in test_filters:
if test_filter(test_reference):
if log is not None:
log.debug(f"Filtering test for {test_reference}, skipping")
skip_test = True
if not skip_test:
filtered_test_references.append(test_reference)
log.info(f"Skipping {len(test_references)-len(filtered_test_references)} out of {len(test_references)} tests.")
test_references = filtered_test_references
if page_size > 0:
slice_start = page_size * page_number
slice_end = page_size * (page_number + 1)
test_references = test_references[slice_start:slice_end]
return test_references
[docs]def main(argv=None):
if argv is None:
argv = sys.argv[1:]
args = arg_parser().parse_args(argv)
try:
run_tests(args)
except Exception as exc:
print(f"ERROR: {exc}", file=sys.stderr)
sys.exit(1)
[docs]def run_tests(args, test_filters=None, log=None):
# Split out argument parsing so we can quickly build other scripts - such as a script
# to run all tool tests for a workflow by just passing in a custom test_filters.
test_filters = test_filters or []
log = log or setup_global_logger(__name__, verbose=args.verbose)
client_test_config_path = args.client_test_config
if client_test_config_path is not None:
log.debug(f"Reading client config path {client_test_config_path}")
with open(client_test_config_path) as f:
client_test_config = yaml.full_load(f)
else:
client_test_config = {}
def get_option(key):
arg_val = getattr(args, key, None)
if arg_val is None and key in client_test_config:
val = client_test_config.get(key)
else:
val = arg_val
return val
output_json_path = get_option("output_json")
galaxy_url = get_option("galaxy_url")
galaxy_interactor_kwds = {
"galaxy_url": galaxy_url,
"master_api_key": get_option("admin_key"),
"api_key": get_option("key"),
"keep_outputs_dir": args.output,
"download_attempts": get_option("download_attempts"),
"download_sleep": get_option("download_sleep"),
"test_data": get_option("test_data"),
}
tool_id = args.tool_id
tool_version = args.tool_version
tools_client_test_config = DictClientTestConfig(client_test_config.get("tools"))
verbose = args.verbose
galaxy_interactor = GalaxyInteractorApi(**galaxy_interactor_kwds)
results = Results(args.suite_name, output_json_path, append=args.append, galaxy_url=galaxy_url)
skip = args.skip
if skip == "executed":
test_filters.append(results.already_executed)
elif skip == "successful":
test_filters.append(results.already_successful)
test_references = build_case_references(
galaxy_interactor,
tool_id=tool_id,
tool_version=tool_version,
test_index=args.test_index,
page_size=args.page_size,
page_number=args.page_number,
test_filters=test_filters,
log=log,
)
log.debug(f"Built {len(test_references)} test references to executed.")
verify_kwds = dict(
client_test_config=tools_client_test_config,
force_path_paste=args.force_path_paste,
skip_with_reference_data=not args.with_reference_data,
quiet=not verbose,
)
test_tools(
galaxy_interactor,
test_references,
results,
log=log,
parallel_tests=args.parallel_tests,
history_per_test_case=args.history_per_test_case,
history_name=args.history_name,
no_history_reuse=args.no_history_reuse,
no_history_cleanup=args.no_history_cleanup,
publish_history=get_option("publish_history"),
verify_kwds=verify_kwds,
)
exceptions = results.test_exceptions
if exceptions:
exception = exceptions[0]
if hasattr(exception, "exception"):
exception = exception.exception
raise exception
[docs]def setup_global_logger(name, log_file=None, verbose=False):
formatter = logging.Formatter("%(asctime)s %(levelname)-5s - %(message)s")
console = logging.StreamHandler()
console.setFormatter(formatter)
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG if verbose else logging.INFO)
logger.addHandler(console)
if not log_file:
# delete = false is chosen here because it is always nice to have a log file
# ready if you need to debug. Not having the "if only I had set a log file"
# moment after the fact.
temp = tempfile.NamedTemporaryFile(prefix="ephemeris_", delete=False)
log_file = temp.name
file_handler = logging.FileHandler(log_file)
logger.addHandler(file_handler)
logger.info(f"Storing log file in: {log_file}")
return logger
[docs]def arg_parser():
parser = argparse.ArgumentParser(description=DESCRIPTION)
parser.add_argument("-u", "--galaxy-url", default="http://localhost:8080", help="Galaxy URL")
parser.add_argument("-k", "--key", default=None, help="Galaxy User API Key")
parser.add_argument("-a", "--admin-key", default=None, help="Galaxy Admin API Key")
parser.add_argument(
"--force_path_paste",
default=False,
action="store_true",
help='This requires Galaxy-side config option "allow_path_paste" enabled. Allows for fetching test data locally. Only for admins.',
)
parser.add_argument("-t", "--tool-id", default=ALL_TOOLS, help="Tool ID")
parser.add_argument(
"--tool-version",
default=None,
help="Tool Version (if tool id supplied). Defaults to just latest version, use * to test all versions",
)
parser.add_argument(
"-i",
"--test-index",
default=ALL_TESTS,
type=int,
help="Tool Test Index (starting at 0) - by default all tests will run.",
)
parser.add_argument("-o", "--output", default=None, help="directory to dump outputs to")
parser.add_argument(
"--append",
default=False,
action="store_true",
help="Extend a test record json (created with --output-json) with additional tests.",
)
skip_group = parser.add_mutually_exclusive_group()
skip_group.add_argument(
"--skip-previously-executed",
dest="skip",
default="no",
action="store_const",
const="executed",
help="When used with --append, skip any test previously executed.",
)
skip_group.add_argument(
"--skip-previously-successful",
dest="skip",
default="no",
action="store_const",
const="successful",
help="When used with --append, skip any test previously executed successfully.",
)
parser.add_argument("-j", "--output-json", default=None, help="output metadata json")
parser.add_argument("--verbose", default=False, action="store_true", help="Verbose logging.")
parser.add_argument("-c", "--client-test-config", default=None, help="Test config YAML to help with client testing")
parser.add_argument("--suite-name", default=DEFAULT_SUITE_NAME, help="Suite name for tool test output")
parser.add_argument("--with-reference-data", dest="with_reference_data", default=False, action="store_true")
parser.add_argument(
"--skip-with-reference-data",
dest="with_reference_data",
action="store_false",
help="Skip tests the Galaxy server believes use data tables or loc files.",
)
history_per_group = parser.add_mutually_exclusive_group()
history_per_group.add_argument(
"--history-per-suite",
dest="history_per_test_case",
default=False,
action="store_false",
help="Create new history per test suite (all tests in same history).",
)
history_per_group.add_argument(
"--history-per-test-case",
dest="history_per_test_case",
action="store_true",
help="Create new history per test case.",
)
history_per_group.add_argument("--history-name", default=None, help="Override default history name")
parser.add_argument(
"--no-history-reuse",
default=False,
action="store_true",
help="Do not reuse histories if a matching one already exists.",
)
parser.add_argument(
"--no-history-cleanup", default=False, action="store_true", help="Perserve histories created for testing."
)
parser.add_argument(
"--publish-history", default=False, action="store_true", help="Publish test history. Useful for CI testing."
)
parser.add_argument("--parallel-tests", default=1, type=int, help="Parallel tests.")
parser.add_argument("--retries", default=0, type=int, help="Retry failed tests.")
parser.add_argument(
"--page-size", default=0, type=int, help="If positive, use pagination and just run one 'page' to tool tests."
)
parser.add_argument(
"--page-number", default=0, type=int, help="If page size is used, run this 'page' of tests - starts with 0."
)
parser.add_argument(
"--download-attempts",
default=1,
type=int,
help="Galaxy may return a transient 500 status code for download if test results are written but not yet accessible.",
)
parser.add_argument(
"--download-sleep",
default=1,
type=int,
help="If download attempts is greater than 1, the amount to sleep between download attempts.",
)
parser.add_argument("--test-data", action="append", help="Add local test data path to search for missing test data")
return parser
if __name__ == "__main__":
main()