diff options
author | monster <monster@ydb.tech> | 2022-07-07 14:41:37 +0300 |
---|---|---|
committer | monster <monster@ydb.tech> | 2022-07-07 14:41:37 +0300 |
commit | 06e5c21a835c0e923506c4ff27929f34e00761c2 (patch) | |
tree | 75efcbc6854ef9bd476eb8bf00cc5c900da436a2 /contrib/python/pytest/py2/_pytest/junitxml.py | |
parent | 03f024c4412e3aa613bb543cf1660176320ba8f4 (diff) | |
download | ydb-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.py | 707 |
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 "" |