path: root/contrib/python/pytest/py2/_pytest/junitxml.py

# -*- coding: utf-8 -*-
""" 
    report test results in JUnit-XML format, 
    for use with Jenkins and build integration servers. 
 
 
Based on initial code from Ross Lawley. 
 
Output conforms to https://github.com/jenkinsci/xunit-plugin/blob/master/ 
src/main/resources/org/jenkinsci/plugins/xunit/types/model/xsd/junit-10.xsd 
""" 
from __future__ import absolute_import 
from __future__ import division 
from __future__ import print_function 
 
import functools 
import os 
import platform
import re 
import sys 
import time 
from datetime import datetime
 
import py 
import six
 
import pytest 
from _pytest import nodes 
from _pytest.config import filename_arg 
 
# Python 2.X and 3.X compatibility 
if sys.version_info[0] < 3: 
    from codecs import open 
 
 
class Junit(py.xml.Namespace): 
    pass 
 
 
# We need to get the subset of the invalid unicode ranges according to 
# XML 1.0 which are valid in this python build.  Hence we calculate 
# this dynamically instead of hardcoding it.  The spec range of valid 
# chars is: Char ::= #x9 | #xA | #xD | [#x20-#xD7FF] | [#xE000-#xFFFD] 
#                    | [#x10000-#x10FFFF] 
_legal_chars = (0x09, 0x0A, 0x0D) 
_legal_ranges = ((0x20, 0x7E), (0x80, 0xD7FF), (0xE000, 0xFFFD), (0x10000, 0x10FFFF)) 
_legal_xml_re = [ 
    u"%s-%s" % (six.unichr(low), six.unichr(high))
    for (low, high) in _legal_ranges 
    if low < sys.maxunicode 
] 
_legal_xml_re = [six.unichr(x) for x in _legal_chars] + _legal_xml_re
illegal_xml_re = re.compile(u"[^%s]" % u"".join(_legal_xml_re))
del _legal_chars 
del _legal_ranges 
del _legal_xml_re 
 
_py_ext_re = re.compile(r"\.py$") 
 
 
def bin_xml_escape(arg): 
    def repl(matchobj): 
        i = ord(matchobj.group()) 
        if i <= 0xFF: 
            return u"#x%02X" % i
        else: 
            return u"#x%04X" % i
 
    return py.xml.raw(illegal_xml_re.sub(repl, py.xml.escape(arg))) 
 
 
def merge_family(left, right):
    result = {}
    for kl, vl in left.items():
        for kr, vr in right.items():
            if not isinstance(vl, list):
                raise TypeError(type(vl))
            result[kl] = vl + vr
    left.update(result)


families = {}
families["_base"] = {"testcase": ["classname", "name"]}
families["_base_legacy"] = {"testcase": ["file", "line", "url"]}

# xUnit 1.x inherits legacy attributes
families["xunit1"] = families["_base"].copy()
merge_family(families["xunit1"], families["_base_legacy"])

# xUnit 2.x uses strict base attributes
families["xunit2"] = families["_base"]


class _NodeReporter(object): 
    def __init__(self, nodeid, xml): 
        self.id = nodeid 
        self.xml = xml 
        self.add_stats = self.xml.add_stats 
        self.family = self.xml.family
        self.duration = 0 
        self.properties = [] 
        self.nodes = [] 
        self.testcase = None 
        self.attrs = {} 
 
    def append(self, node): 
        self.xml.add_stats(type(node).__name__) 
        self.nodes.append(node) 
 
    def add_property(self, name, value): 
        self.properties.append((str(name), bin_xml_escape(value))) 
 
    def add_attribute(self, name, value): 
        self.attrs[str(name)] = bin_xml_escape(value) 
 
    def make_properties_node(self): 
        """Return a Junit node containing custom properties, if any. 
        """ 
        if self.properties: 
            return Junit.properties( 
                [ 
                    Junit.property(name=name, value=value) 
                    for name, value in self.properties 
                ] 
            ) 
        return "" 
 
    def record_testreport(self, testreport): 
        assert not self.testcase 
        names = mangle_test_address(testreport.nodeid) 
        existing_attrs = self.attrs 
        classnames = names[:-1] 
        if self.xml.prefix: 
            classnames.insert(0, self.xml.prefix) 
        attrs = { 
            "classname": ".".join(classnames), 
            "name": bin_xml_escape(names[-1]), 
            "file": testreport.location[0], 
        } 
        if testreport.location[1] is not None: 
            attrs["line"] = testreport.location[1] 
        if hasattr(testreport, "url"): 
            attrs["url"] = testreport.url 
        self.attrs = attrs 
        self.attrs.update(existing_attrs)  # restore any user-defined attributes 
 
        # Preserve legacy testcase behavior
        if self.family == "xunit1":
            return

        # Filter out attributes not permitted by this test family.
        # Including custom attributes because they are not valid here.
        temp_attrs = {}
        for key in self.attrs.keys():
            if key in families[self.family]["testcase"]:
                temp_attrs[key] = self.attrs[key]
        self.attrs = temp_attrs

    def to_xml(self): 
        testcase = Junit.testcase(time="%.3f" % self.duration, **self.attrs)
        testcase.append(self.make_properties_node()) 
        for node in self.nodes: 
            testcase.append(node) 
        return testcase 
 
    def _add_simple(self, kind, message, data=None): 
        data = bin_xml_escape(data) 
        node = kind(data, message=message) 
        self.append(node) 
 
    def write_captured_output(self, report): 
        if not self.xml.log_passing_tests and report.passed:
            return

        content_out = report.capstdout 
        content_log = report.caplog 
        content_err = report.capstderr 
 
        if content_log or content_out: 
            if content_log and self.xml.logging == "system-out": 
                if content_out: 
                    # syncing stdout and the log-output is not done yet. It's 
                    # probably not worth the effort. Therefore, first the captured 
                    # stdout is shown and then the captured logs. 
                    content = "\n".join( 
                        [ 
                            " Captured Stdout ".center(80, "-"), 
                            content_out, 
                            "", 
                            " Captured Log ".center(80, "-"), 
                            content_log, 
                        ] 
                    ) 
                else: 
                    content = content_log 
            else: 
                content = content_out 
 
            if content: 
                tag = getattr(Junit, "system-out") 
                self.append(tag(bin_xml_escape(content))) 
 
        if content_log or content_err: 
            if content_log and self.xml.logging == "system-err": 
                if content_err: 
                    content = "\n".join( 
                        [ 
                            " Captured Stderr ".center(80, "-"), 
                            content_err, 
                            "", 
                            " Captured Log ".center(80, "-"), 
                            content_log, 
                        ] 
                    ) 
                else: 
                    content = content_log 
            else: 
                content = content_err 
 
            if content: 
                tag = getattr(Junit, "system-err") 
                self.append(tag(bin_xml_escape(content))) 
 
    def append_pass(self, report): 
        self.add_stats("passed") 
 
    def append_failure(self, report): 
        # msg = str(report.longrepr.reprtraceback.extraline) 
        if hasattr(report, "wasxfail"): 
            self._add_simple(Junit.skipped, "xfail-marked test passes unexpectedly") 
        else: 
            if hasattr(report.longrepr, "reprcrash"): 
                message = report.longrepr.reprcrash.message 
            elif isinstance(report.longrepr, six.string_types):
                message = report.longrepr 
            else: 
                message = str(report.longrepr) 
            message = bin_xml_escape(message) 
            fail = Junit.failure(message=message) 
            fail.append(bin_xml_escape(report.longrepr)) 
            self.append(fail) 
 
    def append_collect_error(self, report): 
        # msg = str(report.longrepr.reprtraceback.extraline) 
        self.append( 
            Junit.error(bin_xml_escape(report.longrepr), message="collection failure") 
        ) 
 
    def append_collect_skipped(self, report): 
        self._add_simple(Junit.skipped, "collection skipped", report.longrepr) 
 
    def append_error(self, report): 
        if report.when == "teardown":
            msg = "test teardown failure" 
        else: 
            msg = "test setup failure" 
        self._add_simple(Junit.error, msg, report.longrepr) 
 
    def append_skipped(self, report): 
        if hasattr(report, "wasxfail"): 
            xfailreason = report.wasxfail
            if xfailreason.startswith("reason: "):
                xfailreason = xfailreason[8:]
            self.append(
                Junit.skipped(
                    "", type="pytest.xfail", message=bin_xml_escape(xfailreason)
                )
            )
        else: 
            filename, lineno, skipreason = report.longrepr 
            if skipreason.startswith("Skipped: "): 
                skipreason = skipreason[9:] 
            details = "%s:%s: %s" % (filename, lineno, skipreason) 
 
            self.append( 
                Junit.skipped( 
                    bin_xml_escape(details), 
                    type="pytest.skip", 
                    message=bin_xml_escape(skipreason), 
                ) 
            ) 
            self.write_captured_output(report) 
 
    def finalize(self): 
        data = self.to_xml().unicode(indent=0) 
        self.__dict__.clear() 
        self.to_xml = lambda: py.xml.raw(data) 
 
 
def _warn_incompatibility_with_xunit2(request, fixture_name):
    """Emits a PytestWarning about the given fixture being incompatible with newer xunit revisions"""
    from _pytest.warning_types import PytestWarning

    xml = getattr(request.config, "_xml", None)
    if xml is not None and xml.family not in ("xunit1", "legacy"):
        request.node.warn(
            PytestWarning(
                "{fixture_name} is incompatible with junit_family '{family}' (use 'legacy' or 'xunit1')".format(
                    fixture_name=fixture_name, family=xml.family
                )
            )
        )


@pytest.fixture 
def record_property(request): 
    """Add an extra properties the calling test. 
    User properties become part of the test report and are available to the 
    configured reporters, like JUnit XML. 
    The fixture is callable with ``(name, value)``, with value being automatically 
    xml-encoded. 
 
    Example:: 
 
        def test_function(record_property): 
            record_property("example_key", 1) 
    """ 
    _warn_incompatibility_with_xunit2(request, "record_property")
 
    def append_property(name, value): 
        request.node.user_properties.append((name, value)) 
 
    return append_property 
 
 
@pytest.fixture 
def record_xml_attribute(request): 
    """Add extra xml attributes to the tag for the calling test. 
    The fixture is callable with ``(name, value)``, with value being 
    automatically xml-encoded 
    """ 
    from _pytest.warning_types import PytestExperimentalApiWarning
 
    request.node.warn(
        PytestExperimentalApiWarning("record_xml_attribute is an experimental feature")
    )

    _warn_incompatibility_with_xunit2(request, "record_xml_attribute")

    # Declare noop
    def add_attr_noop(name, value):
        pass

    attr_func = add_attr_noop

    xml = getattr(request.config, "_xml", None) 
    if xml is not None: 
        node_reporter = xml.node_reporter(request.node.nodeid) 
        attr_func = node_reporter.add_attribute
 
    return attr_func
 
 
def _check_record_param_type(param, v):
    """Used by record_testsuite_property to check that the given parameter name is of the proper
    type"""
    __tracebackhide__ = True
    if not isinstance(v, six.string_types):
        msg = "{param} parameter needs to be a string, but {g} given"
        raise TypeError(msg.format(param=param, g=type(v).__name__))
 

@pytest.fixture(scope="session")
def record_testsuite_property(request):
    """
    Records a new ``<property>`` tag as child of the root ``<testsuite>``. This is suitable to
    writing global information regarding the entire test suite, and is compatible with ``xunit2`` JUnit family.

    This is a ``session``-scoped fixture which is called with ``(name, value)``. Example:

    .. code-block:: python

        def test_foo(record_testsuite_property):
            record_testsuite_property("ARCH", "PPC")
            record_testsuite_property("STORAGE_TYPE", "CEPH")

    ``name`` must be a string, ``value`` will be converted to a string and properly xml-escaped.
    """

    __tracebackhide__ = True

    def record_func(name, value):
        """noop function in case --junitxml was not passed in the command-line"""
        __tracebackhide__ = True
        _check_record_param_type("name", name)

    xml = getattr(request.config, "_xml", None)
    if xml is not None:
        record_func = xml.add_global_property  # noqa
    return record_func


def pytest_addoption(parser): 
    group = parser.getgroup("terminal reporting") 
    group.addoption( 
        "--junitxml", 
        "--junit-xml", 
        action="store", 
        dest="xmlpath", 
        metavar="path", 
        type=functools.partial(filename_arg, optname="--junitxml"), 
        default=None, 
        help="create junit-xml style report file at given path.", 
    ) 
    group.addoption( 
        "--junitprefix", 
        "--junit-prefix", 
        action="store", 
        metavar="str", 
        default=None, 
        help="prepend prefix to classnames in junit-xml output", 
    ) 
    parser.addini( 
        "junit_suite_name", "Test suite name for JUnit report", default="pytest" 
    ) 
    parser.addini( 
        "junit_logging", 
        "Write captured log messages to JUnit report: " 
        "one of no|system-out|system-err", 
        default="no", 
    )  # choices=['no', 'stdout', 'stderr']) 
    parser.addini(
        "junit_log_passing_tests",
        "Capture log information for passing tests to JUnit report: ",
        type="bool",
        default=True,
    )
    parser.addini(
        "junit_duration_report",
        "Duration time to report: one of total|call",
        default="total",
    )  # choices=['total', 'call'])
    parser.addini(
        "junit_family",
        "Emit XML for schema: one of legacy|xunit1|xunit2",
        default="xunit1",
    )
 
 
def pytest_configure(config): 
    xmlpath = config.option.xmlpath 
    # prevent opening xmllog on slave nodes (xdist) 
    if xmlpath and not hasattr(config, "slaveinput"): 
        config._xml = LogXML( 
            xmlpath, 
            config.option.junitprefix, 
            config.getini("junit_suite_name"), 
            config.getini("junit_logging"), 
            config.getini("junit_duration_report"),
            config.getini("junit_family"),
            config.getini("junit_log_passing_tests"),
        ) 
        config.pluginmanager.register(config._xml) 
 
 
def pytest_unconfigure(config): 
    xml = getattr(config, "_xml", None) 
    if xml: 
        del config._xml 
        config.pluginmanager.unregister(xml) 
 
 
def mangle_test_address(address): 
    path, possible_open_bracket, params = address.partition("[") 
    names = path.split("::") 
    try: 
        names.remove("()") 
    except ValueError: 
        pass 
    # convert file path to dotted path 
    names[0] = names[0].replace(nodes.SEP, ".") 
    names[0] = _py_ext_re.sub("", names[0]) 
    # put any params back 
    names[-1] += possible_open_bracket + params 
    return names 
 
 
class LogXML(object): 
    def __init__(
        self,
        logfile,
        prefix,
        suite_name="pytest",
        logging="no",
        report_duration="total",
        family="xunit1",
        log_passing_tests=True,
    ):
        logfile = os.path.expanduser(os.path.expandvars(logfile)) 
        self.logfile = os.path.normpath(os.path.abspath(logfile)) 
        self.prefix = prefix 
        self.suite_name = suite_name 
        self.logging = logging 
        self.log_passing_tests = log_passing_tests
        self.report_duration = report_duration
        self.family = family
        self.stats = dict.fromkeys(["error", "passed", "failure", "skipped"], 0) 
        self.node_reporters = {}  # nodeid -> _NodeReporter 
        self.node_reporters_ordered = [] 
        self.global_properties = [] 

        # List of reports that failed on call but teardown is pending. 
        self.open_reports = [] 
        self.cnt_double_fail_tests = 0 
 
        # Replaces convenience family with real family
        if self.family == "legacy":
            self.family = "xunit1"

    def finalize(self, report): 
        nodeid = getattr(report, "nodeid", report) 
        # local hack to handle xdist report order 
        slavenode = getattr(report, "node", None) 
        reporter = self.node_reporters.pop((nodeid, slavenode)) 
        if reporter is not None: 
            reporter.finalize() 
 
    def node_reporter(self, report): 
        nodeid = getattr(report, "nodeid", report) 
        # local hack to handle xdist report order 
        slavenode = getattr(report, "node", None) 
 
        key = nodeid, slavenode 
 
        if key in self.node_reporters: 
            # TODO: breasks for --dist=each 
            return self.node_reporters[key] 
 
        reporter = _NodeReporter(nodeid, self) 
 
        self.node_reporters[key] = reporter 
        self.node_reporters_ordered.append(reporter) 
 
        return reporter 
 
    def add_stats(self, key): 
        if key in self.stats: 
            self.stats[key] += 1 
 
    def _opentestcase(self, report): 
        reporter = self.node_reporter(report) 
        reporter.record_testreport(report) 
        return reporter 
 
    def pytest_runtest_logreport(self, report): 
        """handle a setup/call/teardown report, generating the appropriate 
        xml tags as necessary. 
 
        note: due to plugins like xdist, this hook may be called in interlaced 
        order with reports from other nodes. for example: 
 
        usual call order: 
            -> setup node1 
            -> call node1 
            -> teardown node1 
            -> setup node2 
            -> call node2 
            -> teardown node2 
 
        possible call order in xdist: 
            -> setup node1 
            -> call node1 
            -> setup node2 
            -> call node2 
            -> teardown node2 
            -> teardown node1 
        """ 
        close_report = None 
        if report.passed: 
            if report.when == "call":  # ignore setup/teardown 
                reporter = self._opentestcase(report) 
                reporter.append_pass(report) 
        elif report.failed: 
            if report.when == "teardown": 
                # The following vars are needed when xdist plugin is used 
                report_wid = getattr(report, "worker_id", None) 
                report_ii = getattr(report, "item_index", None) 
                close_report = next( 
                    ( 
                        rep 
                        for rep in self.open_reports 
                        if ( 
                            rep.nodeid == report.nodeid 
                            and getattr(rep, "item_index", None) == report_ii 
                            and getattr(rep, "worker_id", None) == report_wid 
                        ) 
                    ), 
                    None, 
                ) 
                if close_report: 
                    # We need to open new testcase in case we have failure in 
                    # call and error in teardown in order to follow junit 
                    # schema 
                    self.finalize(close_report) 
                    self.cnt_double_fail_tests += 1 
            reporter = self._opentestcase(report) 
            if report.when == "call": 
                reporter.append_failure(report) 
                self.open_reports.append(report) 
                if not self.log_passing_tests:
                    reporter.write_captured_output(report)
            else: 
                reporter.append_error(report) 
        elif report.skipped: 
            reporter = self._opentestcase(report) 
            reporter.append_skipped(report) 
        self.update_testcase_duration(report) 
        if report.when == "teardown": 
            reporter = self._opentestcase(report) 
            reporter.write_captured_output(report) 
 
            for propname, propvalue in report.user_properties: 
                reporter.add_property(propname, propvalue) 
 
            self.finalize(report) 
            report_wid = getattr(report, "worker_id", None) 
            report_ii = getattr(report, "item_index", None) 
            close_report = next( 
                ( 
                    rep 
                    for rep in self.open_reports 
                    if ( 
                        rep.nodeid == report.nodeid 
                        and getattr(rep, "item_index", None) == report_ii 
                        and getattr(rep, "worker_id", None) == report_wid 
                    ) 
                ), 
                None, 
            ) 
            if close_report: 
                self.open_reports.remove(close_report) 
 
    def update_testcase_duration(self, report): 
        """accumulates total duration for nodeid from given report and updates 
        the Junit.testcase with the new total if already created. 
        """ 
        if self.report_duration == "total" or report.when == self.report_duration:
            reporter = self.node_reporter(report)
            reporter.duration += getattr(report, "duration", 0.0)
 
    def pytest_collectreport(self, report): 
        if not report.passed: 
            reporter = self._opentestcase(report) 
            if report.failed: 
                reporter.append_collect_error(report) 
            else: 
                reporter.append_collect_skipped(report) 
 
    def pytest_internalerror(self, excrepr): 
        reporter = self.node_reporter("internal") 
        reporter.attrs.update(classname="pytest", name="internal") 
        reporter._add_simple(Junit.error, "internal error", excrepr) 
 
    def pytest_sessionstart(self): 
        self.suite_start_time = time.time() 
 
    def pytest_sessionfinish(self): 
        dirname = os.path.dirname(os.path.abspath(self.logfile)) 
        if not os.path.isdir(dirname): 
            os.makedirs(dirname) 
        logfile = open(self.logfile, "w", encoding="utf-8") 
        suite_stop_time = time.time() 
        suite_time_delta = suite_stop_time - self.suite_start_time 
 
        numtests = ( 
            self.stats["passed"] 
            + self.stats["failure"] 
            + self.stats["skipped"] 
            + self.stats["error"] 
            - self.cnt_double_fail_tests 
        ) 
        logfile.write('<?xml version="1.0" encoding="utf-8"?>') 
 
        suite_node = Junit.testsuite(
            self._get_global_properties_node(),
            [x.to_xml() for x in self.node_reporters_ordered],
            name=self.suite_name,
            errors=self.stats["error"],
            failures=self.stats["failure"],
            skipped=self.stats["skipped"],
            tests=numtests,
            time="%.3f" % suite_time_delta,
            timestamp=datetime.fromtimestamp(self.suite_start_time).isoformat(),
            hostname=platform.node(),
        ) 
        logfile.write(Junit.testsuites([suite_node]).unicode(indent=0))
        logfile.close() 
 
    def pytest_terminal_summary(self, terminalreporter): 
        terminalreporter.write_sep("-", "generated xml file: %s" % (self.logfile)) 
 
    def add_global_property(self, name, value): 
        __tracebackhide__ = True
        _check_record_param_type("name", name)
        self.global_properties.append((name, bin_xml_escape(value)))
 
    def _get_global_properties_node(self): 
        """Return a Junit node containing custom properties, if any. 
        """ 
        if self.global_properties: 
            return Junit.properties( 
                [ 
                    Junit.property(name=name, value=value) 
                    for name, value in self.global_properties 
                ] 
            ) 
        return ""