Source code for _pytest.junitxml

"""
    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
"""
import functools
import os
import platform
import re
import sys
from datetime import datetime
from typing import Dict
from typing import List
from typing import Match
from typing import Optional
from typing import Tuple
from typing import Union

import py

import pytest
from _pytest import deprecated
from _pytest import nodes
from _pytest import timing
from _pytest._code.code import ExceptionRepr
from _pytest.compat import TYPE_CHECKING
from _pytest.config import Config
from _pytest.config import filename_arg
from _pytest.config.argparsing import Parser
from _pytest.fixtures import FixtureRequest
from _pytest.reports import TestReport
from _pytest.store import StoreKey
from _pytest.terminal import TerminalReporter
from _pytest.warnings import _issue_warning_captured

if TYPE_CHECKING:
    from typing import Type


xml_key = StoreKey["LogXML"]()


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 = [
    "{}-{}".format(chr(low), chr(high))
    for (low, high) in _legal_ranges
    if low < sys.maxunicode
]
_legal_xml_re = [chr(x) for x in _legal_chars] + _legal_xml_re
illegal_xml_re = re.compile("[^%s]" % "".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: str) -> py.xml.raw:
    def repl(matchobj: Match[str]) -> str:
        i = ord(matchobj.group())
        if i <= 0xFF:
            return "#x%02X" % i
        else:
            return "#x%04X" % i

    return py.xml.raw(illegal_xml_re.sub(repl, py.xml.escape(arg)))


def merge_family(left, right) -> None:
    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:
    def __init__(self, nodeid: Union[str, TestReport], xml: "LogXML") -> None:
        self.id = nodeid
        self.xml = xml
        self.add_stats = self.xml.add_stats
        self.family = self.xml.family
        self.duration = 0
        self.properties = []  # type: List[Tuple[str, py.xml.raw]]
        self.nodes = []  # type: List[py.xml.Tag]
        self.attrs = {}  # type: Dict[str, Union[str, py.xml.raw]]

    def append(self, node: py.xml.Tag) -> None:
        self.xml.add_stats(type(node).__name__)
        self.nodes.append(node)

    def add_property(self, name: str, value: str) -> None:
        self.properties.append((str(name), bin_xml_escape(value)))

    def add_attribute(self, name: str, value: str) -> None:
        self.attrs[str(name)] = bin_xml_escape(value)

    def make_properties_node(self) -> Union[py.xml.Tag, str]:
        """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: TestReport) -> None:
        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],
        }  # type: Dict[str, Union[str, py.xml.raw]]
        if testreport.location[1] is not None:
            attrs["line"] = str(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) -> py.xml.Tag:
        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: "Type[py.xml.Tag]", message: str, data=None) -> None:
        data = bin_xml_escape(data)
        node = kind(data, message=message)
        self.append(node)

    def write_captured_output(self, report: TestReport) -> None:
        if not self.xml.log_passing_tests and report.passed:
            return

        content_out = report.capstdout
        content_log = report.caplog
        content_err = report.capstderr
        if self.xml.logging == "no":
            return
        content_all = ""
        if self.xml.logging in ["log", "all"]:
            content_all = self._prepare_content(content_log, " Captured Log ")
        if self.xml.logging in ["system-out", "out-err", "all"]:
            content_all += self._prepare_content(content_out, " Captured Out ")
            self._write_content(report, content_all, "system-out")
            content_all = ""
        if self.xml.logging in ["system-err", "out-err", "all"]:
            content_all += self._prepare_content(content_err, " Captured Err ")
            self._write_content(report, content_all, "system-err")
            content_all = ""
        if content_all:
            self._write_content(report, content_all, "system-out")

    def _prepare_content(self, content: str, header: str) -> str:
        return "\n".join([header.center(80, "-"), content, ""])

    def _write_content(self, report: TestReport, content: str, jheader: str) -> None:
        tag = getattr(Junit, jheader)
        self.append(tag(bin_xml_escape(content)))

    def append_pass(self, report: TestReport) -> None:
        self.add_stats("passed")

    def append_failure(self, report: TestReport) -> None:
        # msg = str(report.longrepr.reprtraceback.extraline)
        if hasattr(report, "wasxfail"):
            self._add_simple(Junit.skipped, "xfail-marked test passes unexpectedly")
        else:
            assert report.longrepr is not None
            if getattr(report.longrepr, "reprcrash", None) is not None:
                message = report.longrepr.reprcrash.message
            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: TestReport) -> None:
        # msg = str(report.longrepr.reprtraceback.extraline)
        assert report.longrepr is not None
        self.append(
            Junit.error(bin_xml_escape(report.longrepr), message="collection failure")
        )

    def append_collect_skipped(self, report: TestReport) -> None:
        self._add_simple(Junit.skipped, "collection skipped", report.longrepr)

    def append_error(self, report: TestReport) -> None:
        assert report.longrepr is not None
        if getattr(report.longrepr, "reprcrash", None) is not None:
            reason = report.longrepr.reprcrash.message
        else:
            reason = str(report.longrepr)

        if report.when == "teardown":
            msg = 'failed on teardown with "{}"'.format(reason)
        else:
            msg = 'failed on setup with "{}"'.format(reason)
        self._add_simple(Junit.error, msg, report.longrepr)

    def append_skipped(self, report: TestReport) -> None:
        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:
            assert report.longrepr is not None
            filename, lineno, skipreason = report.longrepr
            if skipreason.startswith("Skipped: "):
                skipreason = skipreason[9:]
            details = "{}:{}: {}".format(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) -> None:
        data = self.to_xml().unicode(indent=0)
        self.__dict__.clear()
        # Type ignored becuase mypy doesn't like overriding a method.
        # Also the return value doesn't match...
        self.to_xml = lambda: py.xml.raw(data)  # type: ignore # noqa: F821


def _warn_incompatibility_with_xunit2(
    request: FixtureRequest, fixture_name: str
) -> None:
    """Emits a PytestWarning about the given fixture being incompatible with newer xunit revisions"""
    from _pytest.warning_types import PytestWarning

    xml = request.config._store.get(xml_key, 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
                )
            )
        )


[docs]@pytest.fixture def record_property(request: FixtureRequest): """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: str, value: object) -> None: request.node.user_properties.append((name, value)) return append_property
@pytest.fixture def record_xml_attribute(request: FixtureRequest): """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: str, value: str) -> None: pass attr_func = add_attr_noop xml = request.config._store.get(xml_key, 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: str, v: str) -> None: """Used by record_testsuite_property to check that the given parameter name is of the proper type""" __tracebackhide__ = True if not isinstance(v, str): msg = "{param} parameter needs to be a string, but {g} given" raise TypeError(msg.format(param=param, g=type(v).__name__))
[docs]@pytest.fixture(scope="session") def record_testsuite_property(request: FixtureRequest): """ 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: str, value: str): """noop function in case --junitxml was not passed in the command-line""" __tracebackhide__ = True _check_record_param_type("name", name) xml = request.config._store.get(xml_key, None) if xml is not None: record_func = xml.add_global_property # noqa return record_func
def pytest_addoption(parser: Parser) -> None: 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|log|system-out|system-err|out-err|all", default="no", ) 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=None ) def pytest_configure(config: Config) -> None: xmlpath = config.option.xmlpath # prevent opening xmllog on worker nodes (xdist) if xmlpath and not hasattr(config, "workerinput"): junit_family = config.getini("junit_family") if not junit_family: _issue_warning_captured(deprecated.JUNIT_XML_DEFAULT_FAMILY, config.hook, 2) junit_family = "xunit1" config._store[xml_key] = LogXML( xmlpath, config.option.junitprefix, config.getini("junit_suite_name"), config.getini("junit_logging"), config.getini("junit_duration_report"), junit_family, config.getini("junit_log_passing_tests"), ) config.pluginmanager.register(config._store[xml_key]) def pytest_unconfigure(config: Config) -> None: xml = config._store.get(xml_key, None) if xml: del config._store[xml_key] config.pluginmanager.unregister(xml) def mangle_test_address(address: str) -> List[str]: 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: def __init__( self, logfile, prefix: Optional[str], suite_name: str = "pytest", logging: str = "no", report_duration: str = "total", family="xunit1", log_passing_tests: bool = True, ) -> None: 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 ) # type: Dict[str, int] self.node_reporters = ( {} ) # type: Dict[Tuple[Union[str, TestReport], object], _NodeReporter] self.node_reporters_ordered = [] # type: List[_NodeReporter] self.global_properties = [] # type: List[Tuple[str, py.xml.raw]] # List of reports that failed on call but teardown is pending. self.open_reports = [] # type: List[TestReport] self.cnt_double_fail_tests = 0 # Replaces convenience family with real family if self.family == "legacy": self.family = "xunit1" def finalize(self, report: TestReport) -> None: nodeid = getattr(report, "nodeid", report) # local hack to handle xdist report order workernode = getattr(report, "node", None) reporter = self.node_reporters.pop((nodeid, workernode)) if reporter is not None: reporter.finalize() def node_reporter(self, report: Union[TestReport, str]) -> _NodeReporter: nodeid = getattr(report, "nodeid", report) # type: Union[str, TestReport] # local hack to handle xdist report order workernode = getattr(report, "node", None) key = nodeid, workernode if key in self.node_reporters: # TODO: breaks 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: str) -> None: if key in self.stats: self.stats[key] += 1 def _opentestcase(self, report: TestReport) -> _NodeReporter: reporter = self.node_reporter(report) reporter.record_testreport(report) return reporter def pytest_runtest_logreport(self, report: TestReport) -> None: """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, str(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: TestReport) -> None: """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: TestReport) -> None: 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: ExceptionRepr) -> None: reporter = self.node_reporter("internal") reporter.attrs.update(classname="pytest", name="internal") reporter._add_simple(Junit.error, "internal error", excrepr) def pytest_sessionstart(self) -> None: self.suite_start_time = timing.time() def pytest_sessionfinish(self) -> None: 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 = timing.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=str(self.stats["error"]), failures=str(self.stats["failure"]), skipped=str(self.stats["skipped"]), tests=str(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) -> None: terminalreporter.write_sep("-", "generated xml file: {}".format(self.logfile)) def add_global_property(self, name: str, value: str) -> None: __tracebackhide__ = True _check_record_param_type("name", name) self.global_properties.append((name, bin_xml_escape(value))) def _get_global_properties_node(self) -> Union[py.xml.Tag, str]: """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 ""