Source code for planemo.galaxy.test.actions

"""Actions related to running and reporting on Galaxy-specific testing."""

import json

import click
from galaxy.util import unicodify

from planemo.exit_codes import (
    EXIT_CODE_GENERIC_FAILURE,
    EXIT_CODE_NO_SUCH_TARGET,
    EXIT_CODE_OK,
)
from planemo.io import (
    info,
    warn,
)
from planemo.reports import (
    allure,
    build_report,
)
from planemo.test.results import get_dict_value
from . import structures as test_structures

NO_XUNIT_REPORT_MESSAGE = (
    "Cannot locate xUnit report [%s] for tests - " "required to build planemo report and summarize " "tests."
)
NO_JSON_REPORT_MESSAGE = (
    "Cannot locate JSON report [%s] for tests - " "required to build planemo report and summarize " "tests."
)
REPORT_NOT_CHANGED = (
    "Galaxy failed to update test report [%s] for tests - " "required to build planemo report and summarize " "tests."
)
NO_TESTS_MESSAGE = "No tests were executed - see Galaxy output for details."
ALL_TESTS_PASSED_MESSAGE = "All %d test(s) executed passed."
PROBLEM_COUNT_MESSAGE = (
    "There were problems with %d test(s) - out of %d " "test(s) executed. See %s for detailed breakdown."
)
GENERIC_PROBLEMS_MESSAGE = "One or more tests failed. See %s for detailed " "breakdown."
GENERIC_TESTS_PASSED_MESSAGE = "No failing tests encountered."
TEST_DATA_UPDATED_MESSAGE = "Test data were updated and tests were rerun."
TEST_DATA_NOT_UPDATED_MESSAGE = "%s Therefore, no test data were updated." % ALL_TESTS_PASSED_MESSAGE


[docs]def handle_reports_and_summary(ctx, structured_data, exit_code=None, kwds=None): """Produce reports and print summary, return 0 if tests passed. If ``exit_code`` is set - use underlying test source for return code and test success determination, otherwise infer from supplied test data. """ if kwds is None: kwds = {} handle_reports(ctx, structured_data, kwds) summary_exit_code = _handle_summary(structured_data, **kwds) return exit_code if exit_code is not None else summary_exit_code
def merge_reports(input_paths, output_path): reports = [] for path in input_paths: with open(path, encoding="utf-8") as f: reports.append(json.load(f)) tests = [] for report in reports: tests.extend(report["tests"]) tests = sorted(tests, key=lambda k: k["id"]) merged_report = {"tests": tests} with open(output_path, mode="w", encoding="utf-8") as out: out.write(unicodify(json.dumps(merged_report)))
[docs]def handle_reports(ctx, structured_data, kwds): """Write reports based on user specified kwds.""" exceptions = [] structured_report_file = kwds.get("test_output_json", None) if structured_report_file: try: with open(structured_report_file, mode="w", encoding="utf-8") as f: f.write(unicodify(json.dumps(structured_data))) except Exception as e: exceptions.append(e) for report_type in ["html", "markdown", "text", "xunit", "junit", "allure"]: try: _handle_test_output_file(ctx, report_type, structured_data, kwds) except Exception as e: exceptions.append(e) if len(exceptions) > 0: raise exceptions[0]
def _handle_test_output_file(ctx, report_type, test_data, kwds): kwd_name = "test_output" if report_type != "html": kwd_name = "test_output_%s" % report_type path = kwds.get(kwd_name, None) if path is None: message = "No file specified for %s, skipping test output." % kwd_name ctx.vlog(message) return if report_type == "allure": file_modication_datatime = kwds.get("file_modication_datatime") allure.write_results(path, test_data, file_modication_datatime=file_modication_datatime) return execution_type = kwds.get("execution_type", "Test") try: contents = build_report.build_report(test_data, report_type=report_type, execution_type=execution_type) except Exception: message = f"Problem producing report file {path} for {kwd_name}" ctx.vlog(message, exception=True) raise try: with open(path, mode="w", encoding="utf-8") as handle: handle.write(unicodify(contents)) except Exception: message = f"Problem writing output file {kwd_name} for {path}" ctx.vlog(message, exception=True) raise def _handle_summary(structured_data, **kwds): summary_dict = get_dict_value("summary", structured_data) num_tests = get_dict_value("num_tests", summary_dict) num_failures = get_dict_value("num_failures", summary_dict) num_errors = get_dict_value("num_errors", summary_dict) num_problems = num_failures + num_errors summary_exit_code = EXIT_CODE_OK if num_problems > 0: summary_exit_code = EXIT_CODE_GENERIC_FAILURE elif num_tests == 0: summary_exit_code = EXIT_CODE_NO_SUCH_TARGET summary_style = kwds.get("summary") if kwds.get("test_data_updated"): info(TEST_DATA_UPDATED_MESSAGE) if summary_style != "none": if num_tests == 0: warn(NO_TESTS_MESSAGE) elif num_problems == 0: if kwds.get("update_test_data") and not kwds.get("test_data_updated"): info(TEST_DATA_NOT_UPDATED_MESSAGE % num_tests) else: info(ALL_TESTS_PASSED_MESSAGE % num_tests) elif num_problems: html_report_file = kwds.get("test_output") message_args = (num_problems, num_tests, html_report_file) message = PROBLEM_COUNT_MESSAGE % message_args warn(message) _summarize_tests_full(structured_data, **kwds) return summary_exit_code def _summarize_tests_full(structured_data, **kwds): tests = get_dict_value("tests", structured_data) for test_case_data in tests: _summarize_test_case(test_case_data, **kwds) def passed(xunit_testcase_el): did_pass = True for child_el in list(xunit_testcase_el): if child_el.tag in ["failure", "error"]: did_pass = False return did_pass def _summarize_test_case(structured_data, **kwds): summary_style = kwds.get("summary") test_id = test_structures.case_id(raw_id=get_dict_value("id", structured_data)) status = get_dict_value("status", get_dict_value("data", structured_data)) if status != "success": state = click.style("failed", bold=True, fg="red") else: state = click.style("passed", bold=True, fg="green") click.echo(test_id.label + ": " + state) if summary_style != "minimal": _print_command_line(structured_data, test_id) def _print_command_line(test, test_id): execution_problem = test.get("execution_problem", None) if execution_problem: click.echo("| command: *could not execute job, no command generated* ") return job = None try: job = test["job"] except (KeyError, IndexError): click.echo("| command: *failed to find job for test object [%s]" % test) return try: command = job["command_line"] except (KeyError, IndexError): click.echo("| command: *failed to find command_line for job object [%s]" % job) return click.echo("| command: %s" % command) __all__ = ( "handle_reports", "handle_reports_and_summary", )