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
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import functools
import os
import re
import sys
import time

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):

# 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(
        if i <= 0xFF:
            return u"#x%02X" % i
            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

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): = nodeid
        self.xml = xml
        self.add_stats = self.xml.add_stats =
        self.duration = 0 = []
        self.nodes = []
        self.testcase = None
        self.attrs = {}

    def append(self, node):

    def add_property(self, name, value):, 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.
          , value=value)
                    for name, value in
        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 == "xunit1":

        # 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[]["testcase"]:
                temp_attrs[key] = self.attrs[key]
        self.attrs = temp_attrs

    def to_xml(self):
        testcase = Junit.testcase(time="%.3f" % self.duration, **self.attrs)
        for node in self.nodes:
        return testcase

    def _add_simple(self, kind, message, data=None):
        data = bin_xml_escape(data)
        node = kind(data, message=message)

    def write_captured_output(self, report):
        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, "-"),
                            " Captured Log ".center(80, "-"),
                    content = content_log
                content = content_out

            if content:
                tag = getattr(Junit, "system-out")

        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, "-"),
                            " Captured Log ".center(80, "-"),
                    content = content_log
                content = content_err

            if content:
                tag = getattr(Junit, "system-err")

    def append_pass(self, report):

    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")
            if hasattr(report.longrepr, "reprcrash"):
                message = report.longrepr.reprcrash.message
            elif isinstance(report.longrepr, six.string_types):
                message = report.longrepr
                message = str(report.longrepr)
            message = bin_xml_escape(message)
            fail = Junit.failure(message=message)

    def append_collect_error(self, report):
        # msg = str(report.longrepr.reprtraceback.extraline)
            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"
            msg = "test setup failure"
        self._add_simple(Junit.error, msg, report.longrepr)

    def append_skipped(self, report):
        if hasattr(report, "wasxfail"):
            self._add_simple(Junit.skipped, "expected test failure", report.wasxfail)
            filename, lineno, skipreason = report.longrepr
            if skipreason.startswith("Skipped: "):
                skipreason = skipreason[9:]
            details = "%s:%s: %s" % (filename, lineno, skipreason)


    def finalize(self):
        data = self.to_xml().unicode(indent=0)
        self.to_xml = lambda: py.xml.raw(data)

[docs]@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) """ 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 PytestWarning request.node.warn(PytestWarning("record_xml_attribute is an experimental feature")) # 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 and != "xunit1": request.node.warn( PytestWarning( "record_xml_attribute is incompatible with junit_family: " "%s (use: legacy|xunit1)" % ) ) elif xml is not None: node_reporter = xml.node_reporter(request.node.nodeid) attr_func = node_reporter.add_attribute return attr_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_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.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", ): 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.report_duration = report_duration = 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 == "legacy": = "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) 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"?>') logfile.write( 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, ).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): self.global_properties.append((str(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 [, value=value) for name, value in self.global_properties ] ) return ""