# Source code for _pytest.junitxml

    # -*- 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 ""