aboutsummaryrefslogtreecommitdiffstats
path: root/contrib/python/pytest/py2/_pytest/junitxml.py
diff options
context:
space:
mode:
authormonster <monster@ydb.tech>2022-07-07 14:41:37 +0300
committermonster <monster@ydb.tech>2022-07-07 14:41:37 +0300
commit06e5c21a835c0e923506c4ff27929f34e00761c2 (patch)
tree75efcbc6854ef9bd476eb8bf00cc5c900da436a2 /contrib/python/pytest/py2/_pytest/junitxml.py
parent03f024c4412e3aa613bb543cf1660176320ba8f4 (diff)
downloadydb-06e5c21a835c0e923506c4ff27929f34e00761c2.tar.gz
fix ya.make
Diffstat (limited to 'contrib/python/pytest/py2/_pytest/junitxml.py')
-rw-r--r--contrib/python/pytest/py2/_pytest/junitxml.py707
1 files changed, 0 insertions, 707 deletions
diff --git a/contrib/python/pytest/py2/_pytest/junitxml.py b/contrib/python/pytest/py2/_pytest/junitxml.py
deleted file mode 100644
index 853dcb7744..0000000000
--- a/contrib/python/pytest/py2/_pytest/junitxml.py
+++ /dev/null
@@ -1,707 +0,0 @@
-# -*- 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 ""