Source code for _pytest.junitxml

# mypy: allow-untyped-defs
"""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 annotations

from collections.abc import Callable
from datetime import datetime
from datetime import timezone
import functools
import os
import platform
import re
import xml.etree.ElementTree as ET

from _pytest import nodes
from _pytest import timing
from _pytest._code.code import ExceptionRepr
from _pytest._code.code import ReprFileLocation
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.stash import StashKey
from _pytest.terminal import TerminalReporter
import pytest


xml_key = StashKey["LogXML"]()


def bin_xml_escape(arg: object) -> str:
    r"""Visually escape invalid XML characters.

    For example, transforms
        'hello\aworld\b'
    into
        'hello#x07world#x08'
    Note that the #xABs are *not* XML escapes - missing the ampersand &#xAB.
    The idea is to escape visually for the user rather than for XML itself.
    """

    def repl(matchobj: re.Match[str]) -> str:
        i = ord(matchobj.group())
        if i <= 0xFF:
            return f"#x{i:02X}"
        else:
            return f"#x{i:04X}"

    # The spec range of valid chars is:
    # Char ::= #x9 | #xA | #xD | [#x20-#xD7FF] | [#xE000-#xFFFD] | [#x10000-#x10FFFF]
    # For an unknown(?) reason, we disallow #x7F (DEL) as well.
    illegal_xml_re = (
        "[^\u0009\u000a\u000d\u0020-\u007e\u0080-\ud7ff\ue000-\ufffd\u10000-\u10ffff]"
    )
    return re.sub(illegal_xml_re, repl, str(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 = {  # pylint: disable=dict-init-mutate
    "_base": {"testcase": ["classname", "name"]},
    "_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: 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.0
        self.properties: list[tuple[str, str]] = []
        self.nodes: list[ET.Element] = []
        self.attrs: dict[str, str] = {}

    def append(self, node: ET.Element) -> None:
        self.xml.add_stats(node.tag)
        self.nodes.append(node)

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

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

    def make_properties_node(self) -> ET.Element | None:
        """Return a Junit node containing custom properties, if any."""
        if self.properties:
            properties = ET.Element("properties")
            for name, value in self.properties:
                properties.append(ET.Element("property", name=name, value=value))
            return properties
        return None

    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: dict[str, str] = {
            "classname": ".".join(classnames),
            "name": bin_xml_escape(names[-1]),
            "file": testreport.location[0],
        }
        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:
            if key in families[self.family]["testcase"]:
                temp_attrs[key] = self.attrs[key]
        self.attrs = temp_attrs

    def to_xml(self) -> ET.Element:
        testcase = ET.Element("testcase", self.attrs, time=f"{self.duration:.3f}")
        properties = self.make_properties_node()
        if properties is not None:
            testcase.append(properties)
        testcase.extend(self.nodes)
        return testcase

    def _add_simple(self, tag: str, message: str, data: str | None = None) -> None:
        node = ET.Element(tag, message=message)
        node.text = bin_xml_escape(data)
        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 = ET.Element(jheader)
        tag.text = bin_xml_escape(content)
        self.append(tag)

    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("skipped", "xfail-marked test passes unexpectedly")
        else:
            assert report.longrepr is not None
            reprcrash: ReprFileLocation | None = getattr(
                report.longrepr, "reprcrash", None
            )
            if reprcrash is not None:
                message = reprcrash.message
            else:
                message = str(report.longrepr)
            message = bin_xml_escape(message)
            self._add_simple("failure", message, str(report.longrepr))

    def append_collect_error(self, report: TestReport) -> None:
        # msg = str(report.longrepr.reprtraceback.extraline)
        assert report.longrepr is not None
        self._add_simple("error", "collection failure", str(report.longrepr))

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

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

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

    def append_skipped(self, report: TestReport) -> None:
        if hasattr(report, "wasxfail"):
            xfailreason = report.wasxfail
            if xfailreason.startswith("reason: "):
                xfailreason = xfailreason[8:]
            xfailreason = bin_xml_escape(xfailreason)
            skipped = ET.Element("skipped", type="pytest.xfail", message=xfailreason)
            self.append(skipped)
        else:
            assert isinstance(report.longrepr, tuple)
            filename, lineno, skipreason = report.longrepr
            if skipreason.startswith("Skipped: "):
                skipreason = skipreason[9:]
            details = f"{filename}:{lineno}: {skipreason}"

            skipped = ET.Element(
                "skipped", type="pytest.skip", message=bin_xml_escape(skipreason)
            )
            skipped.text = bin_xml_escape(details)
            self.append(skipped)
            self.write_captured_output(report)

    def finalize(self) -> None:
        data = self.to_xml()
        self.__dict__.clear()
        # Type ignored because mypy doesn't like overriding a method.
        # Also the return value doesn't match...
        self.to_xml = lambda: data  # type: ignore[method-assign]


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

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


[docs] @pytest.fixture def record_property(request: FixtureRequest) -> Callable[[str, object], None]: """Add extra properties to 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``. The value is 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) -> Callable[[str, object], None]: """Add extra xml attributes to the tag for the calling test. The fixture is callable with ``name, value``. The value is 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: object) -> None: pass attr_func = add_attr_noop xml = request.config.stash.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" # type: ignore[unreachable] raise TypeError(msg.format(param=param, g=type(v).__name__))
[docs] @pytest.fixture(scope="session") def record_testsuite_property(request: FixtureRequest) -> Callable[[str, object], None]: """Record 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") :param name: The property name. :param value: The property value. Will be converted to a string. .. warning:: Currently this fixture **does not work** with the `pytest-xdist <https://github.com/pytest-dev/pytest-xdist>`__ plugin. See :issue:`7767` for details. """ __tracebackhide__ = True def record_func(name: str, value: object) -> None: """No-op function in case --junit-xml was not passed in the command-line.""" __tracebackhide__ = True _check_record_param_type("name", name) xml = request.config.stash.get(xml_key, None) if xml is not None: record_func = xml.add_global_property 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="xunit2", ) 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") config.stash[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.stash[xml_key]) def pytest_unconfigure(config: Config) -> None: xml = config.stash.get(xml_key, None) if xml: del config.stash[xml_key] config.pluginmanager.unregister(xml) def mangle_test_address(address: str) -> list[str]: path, possible_open_bracket, params = address.partition("[") names = path.split("::") # Convert file path to dotted path. names[0] = names[0].replace(nodes.SEP, ".") names[0] = re.sub(r"\.py$", "", names[0]) # Put any params back. names[-1] += possible_open_bracket + params return names class LogXML: def __init__( self, logfile, prefix: str | None, 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[str, int] = dict.fromkeys( ["error", "passed", "failure", "skipped"], 0 ) self.node_reporters: dict[tuple[str | TestReport, object], _NodeReporter] = {} self.node_reporters_ordered: list[_NodeReporter] = [] self.global_properties: list[tuple[str, str]] = [] # List of reports that failed on call but teardown is pending. self.open_reports: 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)) for propname, propvalue in report.user_properties: reporter.add_property(propname, str(propvalue)) if reporter is not None: reporter.finalize() def node_reporter(self, report: TestReport | str) -> _NodeReporter: nodeid: str | TestReport = getattr(report, "nodeid", report) # 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) 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: """Accumulate total duration for nodeid from given report and update the Junit.testcase with the new total if already created.""" if self.report_duration in {"total", report.when}: 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("error", "internal error", str(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)) # exist_ok avoids filesystem race conditions between checking path existence and requesting creation os.makedirs(dirname, exist_ok=True) with open(self.logfile, "w", encoding="utf-8") as logfile: 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 = ET.Element( "testsuite", name=self.suite_name, errors=str(self.stats["error"]), failures=str(self.stats["failure"]), skipped=str(self.stats["skipped"]), tests=str(numtests), time=f"{suite_time_delta:.3f}", timestamp=datetime.fromtimestamp(self.suite_start_time, timezone.utc) .astimezone() .isoformat(), hostname=platform.node(), ) global_properties = self._get_global_properties_node() if global_properties is not None: suite_node.append(global_properties) for node_reporter in self.node_reporters_ordered: suite_node.append(node_reporter.to_xml()) testsuites = ET.Element("testsuites") testsuites.append(suite_node) logfile.write(ET.tostring(testsuites, encoding="unicode")) def pytest_terminal_summary(self, terminalreporter: TerminalReporter) -> None: terminalreporter.write_sep("-", f"generated xml file: {self.logfile}") def add_global_property(self, name: str, value: object) -> None: __tracebackhide__ = True _check_record_param_type("name", name) self.global_properties.append((name, bin_xml_escape(value))) def _get_global_properties_node(self) -> ET.Element | None: """Return a Junit node containing custom properties, if any.""" if self.global_properties: properties = ET.Element("properties") for name, value in self.global_properties: properties.append(ET.Element("property", name=name, value=value)) return properties return None