intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 707 insertions, 0 deletions

diff --git a/contrib/python/pytest/py2/_pytest/junitxml.py b/contrib/python/pytest/py2/_pytest/junitxml.py
new file mode 100644
index 0000000000..853dcb7744
--- /dev/null
+++ b/contrib/python/pytest/py2/_pytest/junitxml.py
@@ -0,0 +1,707 @@
+# -*- 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 ""