lit/lit/reports.py

import abc
import base64
import datetime
import itertools
import json
import os
import tempfile

from xml.sax.saxutils import quoteattr as quo

import lit.Test


def by_suite_and_test_path(test):
    # Suite names are not necessarily unique.  Include object identity in sort
    # key to avoid mixing tests of different suites.
    return (test.suite.name, id(test.suite), test.path_in_suite)


class Report(object):
    def __init__(self, output_file):
        self.output_file = output_file
        # Set by the option parser later.
        self.use_unique_output_file_name = False

    def write_results(self, tests, elapsed):
        if self.use_unique_output_file_name:
            filename, ext = os.path.splitext(os.path.basename(self.output_file))
            fd, _ = tempfile.mkstemp(
                suffix=ext, prefix=f"{filename}.", dir=os.path.dirname(self.output_file)
            )
            report_file = os.fdopen(fd, "w")
        else:
            # Overwrite if the results already exist.
            report_file = open(self.output_file, "w")

        with report_file:
            self._write_results_to_file(tests, elapsed, report_file)

    @abc.abstractmethod
    def _write_results_to_file(self, tests, elapsed, file):
        """Write test results to the file object "file"."""
        pass


class JsonReport(Report):
    def _write_results_to_file(self, tests, elapsed, file):
        unexecuted_codes = {lit.Test.EXCLUDED, lit.Test.SKIPPED}
        tests = [t for t in tests if t.result.code not in unexecuted_codes]
        # Construct the data we will write.
        data = {}
        # Encode the current lit version as a schema version.
        data["__version__"] = lit.__versioninfo__
        data["elapsed"] = elapsed
        # FIXME: Record some information on the lit configuration used?
        # FIXME: Record information from the individual test suites?

        # Encode the tests.
        data["tests"] = tests_data = []
        for test in tests:
            test_data = {
                "name": test.getFullName(),
                "code": test.result.code.name,
                "output": test.result.output,
                "elapsed": test.result.elapsed,
            }

            # Add test metrics, if present.
            if test.result.metrics:
                test_data["metrics"] = metrics_data = {}
                for key, value in test.result.metrics.items():
                    metrics_data[key] = value.todata()

            # Report micro-tests separately, if present
            if test.result.microResults:
                for key, micro_test in test.result.microResults.items():
                    # Expand parent test name with micro test name
                    parent_name = test.getFullName()
                    micro_full_name = parent_name + ":" + key

                    micro_test_data = {
                        "name": micro_full_name,
                        "code": micro_test.code.name,
                        "output": micro_test.output,
                        "elapsed": micro_test.elapsed,
                    }
                    if micro_test.metrics:
                        micro_test_data["metrics"] = micro_metrics_data = {}
                        for key, value in micro_test.metrics.items():
                            micro_metrics_data[key] = value.todata()

                    tests_data.append(micro_test_data)

            tests_data.append(test_data)

        json.dump(data, file, indent=2, sort_keys=True)
        file.write("\n")


_invalid_xml_chars_dict = {
    c: None for c in range(32) if chr(c) not in ("\t", "\n", "\r")
}


def remove_invalid_xml_chars(s):
    # According to the XML 1.0 spec, control characters other than
    # \t,\r, and \n are not permitted anywhere in the document
    # (https://www.w3.org/TR/xml/#charsets) and therefore this function
    # removes them to produce a valid XML document.
    #
    # Note: In XML 1.1 only \0 is illegal (https://www.w3.org/TR/xml11/#charsets)
    # but lit currently produces XML 1.0 output.
    return s.translate(_invalid_xml_chars_dict)


class XunitReport(Report):
    skipped_codes = {lit.Test.EXCLUDED, lit.Test.SKIPPED, lit.Test.UNSUPPORTED}

    def _write_results_to_file(self, tests, elapsed, file):
        tests.sort(key=by_suite_and_test_path)
        tests_by_suite = itertools.groupby(tests, lambda t: t.suite)

        file.write('<?xml version="1.0" encoding="UTF-8"?>\n')
        file.write('<testsuites time="{time:.2f}">\n'.format(time=elapsed))
        for suite, test_iter in tests_by_suite:
            self._write_testsuite(file, suite, list(test_iter))
        file.write("</testsuites>\n")

    def _write_testsuite(self, file, suite, tests):
        skipped = 0
        failures = 0
        time = 0.0

        for t in tests:
            if t.result.code in self.skipped_codes:
                skipped += 1
            if t.isFailure():
                failures += 1
            time += t.result.elapsed or 0.0

        name = suite.config.name.replace(".", "-")
        file.write(
            f'<testsuite name={quo(name)} tests="{len(tests)}" failures="{failures}" skipped="{skipped}" time="{time:.2f}">\n'
        )
        for test in tests:
            self._write_test(file, test, name)
        file.write("</testsuite>\n")

    def _write_test(self, file, test, suite_name):
        path = "/".join(test.path_in_suite[:-1]).replace(".", "_")
        class_name = f"{suite_name}.{path or suite_name}"
        name = test.path_in_suite[-1]
        time = test.result.elapsed or 0.0
        file.write(
            f'<testcase classname={quo(class_name)} name={quo(name)} time="{time:.2f}"'
        )

        if test.isFailure():
            file.write(">\n  <failure><![CDATA[")
            # In the unlikely case that the output contains the CDATA
            # terminator we wrap it by creating a new CDATA block.
            output = test.result.output.replace("]]>", "]]]]><![CDATA[>")
            if isinstance(output, bytes):
                output = output.decode("utf-8", "ignore")

            # Failing test  output sometimes contains control characters like
            # \x1b (e.g. if there was some -fcolor-diagnostics output) which are
            # not allowed inside XML files.
            # This causes problems with CI systems: for example, the Jenkins
            # JUnit XML will throw an exception when ecountering those
            # characters and similar problems also occur with GitLab CI.
            output = remove_invalid_xml_chars(output)
            file.write(output)
            file.write("]]></failure>\n</testcase>\n")
        elif test.result.code in self.skipped_codes:
            reason = self._get_skip_reason(test)
            file.write(f">\n  <skipped message={quo(reason)}/>\n</testcase>\n")
        else:
            file.write("/>\n")

    def _get_skip_reason(self, test):
        code = test.result.code
        if code == lit.Test.EXCLUDED:
            return "Test not selected (--filter, --max-tests)"
        if code == lit.Test.SKIPPED:
            return "User interrupt"

        assert code == lit.Test.UNSUPPORTED
        features = test.getMissingRequiredFeatures()
        if features:
            return "Missing required feature(s): " + ", ".join(features)
        return "Unsupported configuration"


def gen_resultdb_test_entry(
    test_name, start_time, elapsed_time, test_output, result_code, is_expected
):
    test_data = {
        "testId": test_name,
        "start_time": datetime.datetime.fromtimestamp(start_time).isoformat() + "Z",
        "duration": "%.9fs" % elapsed_time,
        "summary_html": '<p><text-artifact artifact-id="artifact-content-in-request"></p>',
        "artifacts": {
            "artifact-content-in-request": {
                "contents": base64.b64encode(test_output.encode("utf-8")).decode(
                    "utf-8"
                ),
            },
        },
        "expected": is_expected,
    }
    if (
        result_code == lit.Test.PASS
        or result_code == lit.Test.XPASS
        or result_code == lit.Test.FLAKYPASS
    ):
        test_data["status"] = "PASS"
    elif result_code == lit.Test.FAIL or result_code == lit.Test.XFAIL:
        test_data["status"] = "FAIL"
    elif (
        result_code == lit.Test.UNSUPPORTED
        or result_code == lit.Test.SKIPPED
        or result_code == lit.Test.EXCLUDED
    ):
        test_data["status"] = "SKIP"
    elif result_code == lit.Test.UNRESOLVED or result_code == lit.Test.TIMEOUT:
        test_data["status"] = "ABORT"
    return test_data


class ResultDBReport(Report):
    def _write_results_to_file(self, tests, elapsed, file):
        unexecuted_codes = {lit.Test.EXCLUDED, lit.Test.SKIPPED}
        tests = [t for t in tests if t.result.code not in unexecuted_codes]
        data = {}
        data["__version__"] = lit.__versioninfo__
        data["elapsed"] = elapsed
        # Encode the tests.
        data["tests"] = tests_data = []
        for test in tests:
            tests_data.append(
                gen_resultdb_test_entry(
                    test_name=test.getFullName(),
                    start_time=test.result.start,
                    elapsed_time=test.result.elapsed,
                    test_output=test.result.output,
                    result_code=test.result.code,
                    is_expected=not test.result.code.isFailure,
                )
            )
            if test.result.microResults:
                for key, micro_test in test.result.microResults.items():
                    # Expand parent test name with micro test name
                    parent_name = test.getFullName()
                    micro_full_name = parent_name + ":" + key + "microres"
                    tests_data.append(
                        gen_resultdb_test_entry(
                            test_name=micro_full_name,
                            start_time=micro_test.start
                            if micro_test.start
                            else test.result.start,
                            elapsed_time=micro_test.elapsed
                            if micro_test.elapsed
                            else test.result.elapsed,
                            test_output=micro_test.output,
                            result_code=micro_test.code,
                            is_expected=not micro_test.code.isFailure,
                        )
                    )

        json.dump(data, file, indent=2, sort_keys=True)
        file.write("\n")


class TimeTraceReport(Report):
    skipped_codes = {lit.Test.EXCLUDED, lit.Test.SKIPPED, lit.Test.UNSUPPORTED}

    def _write_results_to_file(self, tests, elapsed, file):
        # Find when first test started so we can make start times relative.
        first_start_time = min([t.result.start for t in tests])
        events = [
            self._get_test_event(x, first_start_time)
            for x in tests
            if x.result.code not in self.skipped_codes
        ]

        json_data = {"traceEvents": events}

        json.dump(json_data, time_trace_file, indent=2, sort_keys=True)

    def _get_test_event(self, test, first_start_time):
        test_name = test.getFullName()
        elapsed_time = test.result.elapsed or 0.0
        start_time = test.result.start - first_start_time if test.result.start else 0.0
        pid = test.result.pid or 0
        return {
            "pid": pid,
            "tid": 1,
            "ph": "X",
            "ts": int(start_time * 1000000.0),
            "dur": int(elapsed_time * 1000000.0),
            "name": test_name,
        }