704 lines
25 KiB
Python
704 lines
25 KiB
Python
"""
|
|
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
|
|
"""
|
|
import functools
|
|
import os
|
|
import platform
|
|
import re
|
|
import sys
|
|
from datetime import datetime
|
|
from typing import Dict
|
|
from typing import List
|
|
from typing import Optional
|
|
from typing import Tuple
|
|
from typing import Union
|
|
|
|
import py
|
|
|
|
import pytest
|
|
from _pytest import deprecated
|
|
from _pytest import nodes
|
|
from _pytest import timing
|
|
from _pytest.compat import TYPE_CHECKING
|
|
from _pytest.config import Config
|
|
from _pytest.config import filename_arg
|
|
from _pytest.config.argparsing import Parser
|
|
from _pytest.fixtures import FixtureRequest
|
|
from _pytest.reports import TestReport
|
|
from _pytest.store import StoreKey
|
|
from _pytest.terminal import TerminalReporter
|
|
from _pytest.warnings import _issue_warning_captured
|
|
|
|
if TYPE_CHECKING:
|
|
from typing import Type
|
|
|
|
|
|
xml_key = StoreKey["LogXML"]()
|
|
|
|
|
|
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 = [
|
|
"{}-{}".format(chr(low), chr(high))
|
|
for (low, high) in _legal_ranges
|
|
if low < sys.maxunicode
|
|
]
|
|
_legal_xml_re = [chr(x) for x in _legal_chars] + _legal_xml_re
|
|
illegal_xml_re = re.compile("[^%s]" % "".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: str) -> py.xml.raw:
|
|
def repl(matchobj: "re.Match[str]") -> str:
|
|
i = ord(matchobj.group())
|
|
if i <= 0xFF:
|
|
return "#x%02X" % i
|
|
else:
|
|
return "#x%04X" % i
|
|
|
|
return py.xml.raw(illegal_xml_re.sub(repl, py.xml.escape(arg)))
|
|
|
|
|
|
def merge_family(left, right) -> None:
|
|
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:
|
|
def __init__(self, nodeid: Union[str, TestReport], xml: "LogXML") -> None:
|
|
self.id = nodeid
|
|
self.xml = xml
|
|
self.add_stats = self.xml.add_stats
|
|
self.family = self.xml.family
|
|
self.duration = 0
|
|
self.properties = [] # type: List[Tuple[str, py.xml.raw]]
|
|
self.nodes = [] # type: List[py.xml.Tag]
|
|
self.attrs = {} # type: Dict[str, Union[str, py.xml.raw]]
|
|
|
|
def append(self, node: py.xml.Tag) -> None:
|
|
self.xml.add_stats(type(node).__name__)
|
|
self.nodes.append(node)
|
|
|
|
def add_property(self, name: str, value: str) -> None:
|
|
self.properties.append((str(name), bin_xml_escape(value)))
|
|
|
|
def add_attribute(self, name: str, value: str) -> None:
|
|
self.attrs[str(name)] = bin_xml_escape(value)
|
|
|
|
def make_properties_node(self) -> Union[py.xml.Tag, str]:
|
|
"""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: TestReport) -> None:
|
|
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],
|
|
} # type: Dict[str, Union[str, py.xml.raw]]
|
|
if testreport.location[1] is not None:
|
|
attrs["line"] = str(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) -> py.xml.Tag:
|
|
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: "Type[py.xml.Tag]", message: str, data=None) -> None:
|
|
data = bin_xml_escape(data)
|
|
node = kind(data, message=message)
|
|
self.append(node)
|
|
|
|
def write_captured_output(self, report: TestReport) -> None:
|
|
if not self.xml.log_passing_tests and report.passed:
|
|
return
|
|
|
|
content_out = report.capstdout
|
|
content_log = report.caplog
|
|
content_err = report.capstderr
|
|
if self.xml.logging == "no":
|
|
return
|
|
content_all = ""
|
|
if self.xml.logging in ["log", "all"]:
|
|
content_all = self._prepare_content(content_log, " Captured Log ")
|
|
if self.xml.logging in ["system-out", "out-err", "all"]:
|
|
content_all += self._prepare_content(content_out, " Captured Out ")
|
|
self._write_content(report, content_all, "system-out")
|
|
content_all = ""
|
|
if self.xml.logging in ["system-err", "out-err", "all"]:
|
|
content_all += self._prepare_content(content_err, " Captured Err ")
|
|
self._write_content(report, content_all, "system-err")
|
|
content_all = ""
|
|
if content_all:
|
|
self._write_content(report, content_all, "system-out")
|
|
|
|
def _prepare_content(self, content: str, header: str) -> str:
|
|
return "\n".join([header.center(80, "-"), content, ""])
|
|
|
|
def _write_content(self, report: TestReport, content: str, jheader: str) -> None:
|
|
tag = getattr(Junit, jheader)
|
|
self.append(tag(bin_xml_escape(content)))
|
|
|
|
def append_pass(self, report: TestReport) -> None:
|
|
self.add_stats("passed")
|
|
|
|
def append_failure(self, report: TestReport) -> None:
|
|
# msg = str(report.longrepr.reprtraceback.extraline)
|
|
if hasattr(report, "wasxfail"):
|
|
self._add_simple(Junit.skipped, "xfail-marked test passes unexpectedly")
|
|
else:
|
|
assert report.longrepr is not None
|
|
if getattr(report.longrepr, "reprcrash", None) is not None:
|
|
message = report.longrepr.reprcrash.message
|
|
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: TestReport) -> None:
|
|
# msg = str(report.longrepr.reprtraceback.extraline)
|
|
assert report.longrepr is not None
|
|
self.append(
|
|
Junit.error(bin_xml_escape(report.longrepr), message="collection failure")
|
|
)
|
|
|
|
def append_collect_skipped(self, report: TestReport) -> None:
|
|
self._add_simple(Junit.skipped, "collection skipped", report.longrepr)
|
|
|
|
def append_error(self, report: TestReport) -> None:
|
|
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: TestReport) -> None:
|
|
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:
|
|
assert report.longrepr is not None
|
|
filename, lineno, skipreason = report.longrepr
|
|
if skipreason.startswith("Skipped: "):
|
|
skipreason = skipreason[9:]
|
|
details = "{}:{}: {}".format(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) -> None:
|
|
data = self.to_xml().unicode(indent=0)
|
|
self.__dict__.clear()
|
|
# Type ignored becuase mypy doesn't like overriding a method.
|
|
# Also the return value doesn't match...
|
|
self.to_xml = lambda: py.xml.raw(data) # type: ignore # noqa: F821
|
|
|
|
|
|
def _warn_incompatibility_with_xunit2(
|
|
request: FixtureRequest, fixture_name: str
|
|
) -> None:
|
|
"""Emits a PytestWarning about the given fixture being incompatible with newer xunit revisions"""
|
|
from _pytest.warning_types import PytestWarning
|
|
|
|
xml = request.config._store.get(xml_key, 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: FixtureRequest):
|
|
"""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: str, value: object) -> None:
|
|
request.node.user_properties.append((name, value))
|
|
|
|
return append_property
|
|
|
|
|
|
@pytest.fixture
|
|
def record_xml_attribute(request: FixtureRequest):
|
|
"""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: str, value: str) -> None:
|
|
pass
|
|
|
|
attr_func = add_attr_noop
|
|
|
|
xml = request.config._store.get(xml_key, 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: str, v: str) -> None:
|
|
"""Used by record_testsuite_property to check that the given parameter name is of the proper
|
|
type"""
|
|
__tracebackhide__ = True
|
|
if not isinstance(v, str):
|
|
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: FixtureRequest):
|
|
"""
|
|
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: str, value: str):
|
|
"""noop function in case --junitxml was not passed in the command-line"""
|
|
__tracebackhide__ = True
|
|
_check_record_param_type("name", name)
|
|
|
|
xml = request.config._store.get(xml_key, None)
|
|
if xml is not None:
|
|
record_func = xml.add_global_property # noqa
|
|
return record_func
|
|
|
|
|
|
def pytest_addoption(parser: Parser) -> None:
|
|
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|log|system-out|system-err|out-err|all",
|
|
default="no",
|
|
)
|
|
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=None
|
|
)
|
|
|
|
|
|
def pytest_configure(config: Config) -> None:
|
|
xmlpath = config.option.xmlpath
|
|
# prevent opening xmllog on worker nodes (xdist)
|
|
if xmlpath and not hasattr(config, "workerinput"):
|
|
junit_family = config.getini("junit_family")
|
|
if not junit_family:
|
|
_issue_warning_captured(deprecated.JUNIT_XML_DEFAULT_FAMILY, config.hook, 2)
|
|
junit_family = "xunit1"
|
|
config._store[xml_key] = LogXML(
|
|
xmlpath,
|
|
config.option.junitprefix,
|
|
config.getini("junit_suite_name"),
|
|
config.getini("junit_logging"),
|
|
config.getini("junit_duration_report"),
|
|
junit_family,
|
|
config.getini("junit_log_passing_tests"),
|
|
)
|
|
config.pluginmanager.register(config._store[xml_key])
|
|
|
|
|
|
def pytest_unconfigure(config: Config) -> None:
|
|
xml = config._store.get(xml_key, None)
|
|
if xml:
|
|
del config._store[xml_key]
|
|
config.pluginmanager.unregister(xml)
|
|
|
|
|
|
def mangle_test_address(address: str) -> List[str]:
|
|
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:
|
|
def __init__(
|
|
self,
|
|
logfile,
|
|
prefix: Optional[str],
|
|
suite_name: str = "pytest",
|
|
logging: str = "no",
|
|
report_duration: str = "total",
|
|
family="xunit1",
|
|
log_passing_tests: bool = True,
|
|
) -> None:
|
|
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
|
|
) # type: Dict[str, int]
|
|
self.node_reporters = (
|
|
{}
|
|
) # type: Dict[Tuple[Union[str, TestReport], object], _NodeReporter]
|
|
self.node_reporters_ordered = [] # type: List[_NodeReporter]
|
|
self.global_properties = [] # type: List[Tuple[str, py.xml.raw]]
|
|
|
|
# List of reports that failed on call but teardown is pending.
|
|
self.open_reports = [] # type: List[TestReport]
|
|
self.cnt_double_fail_tests = 0
|
|
|
|
# Replaces convenience family with real family
|
|
if self.family == "legacy":
|
|
self.family = "xunit1"
|
|
|
|
def finalize(self, report: TestReport) -> None:
|
|
nodeid = getattr(report, "nodeid", report)
|
|
# local hack to handle xdist report order
|
|
workernode = getattr(report, "node", None)
|
|
reporter = self.node_reporters.pop((nodeid, workernode))
|
|
if reporter is not None:
|
|
reporter.finalize()
|
|
|
|
def node_reporter(self, report: Union[TestReport, str]) -> _NodeReporter:
|
|
nodeid = getattr(report, "nodeid", report) # type: Union[str, TestReport]
|
|
# local hack to handle xdist report order
|
|
workernode = getattr(report, "node", None)
|
|
|
|
key = nodeid, workernode
|
|
|
|
if key in self.node_reporters:
|
|
# TODO: breaks 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: str) -> None:
|
|
if key in self.stats:
|
|
self.stats[key] += 1
|
|
|
|
def _opentestcase(self, report: TestReport) -> _NodeReporter:
|
|
reporter = self.node_reporter(report)
|
|
reporter.record_testreport(report)
|
|
return reporter
|
|
|
|
def pytest_runtest_logreport(self, report: TestReport) -> None:
|
|
"""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, str(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: TestReport) -> None:
|
|
"""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: TestReport) -> None:
|
|
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) -> None:
|
|
reporter = self.node_reporter("internal")
|
|
reporter.attrs.update(classname="pytest", name="internal")
|
|
reporter._add_simple(Junit.error, "internal error", excrepr)
|
|
|
|
def pytest_sessionstart(self) -> None:
|
|
self.suite_start_time = timing.time()
|
|
|
|
def pytest_sessionfinish(self) -> None:
|
|
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 = timing.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=str(self.stats["error"]),
|
|
failures=str(self.stats["failure"]),
|
|
skipped=str(self.stats["skipped"]),
|
|
tests=str(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) -> None:
|
|
terminalreporter.write_sep("-", "generated xml file: {}".format(self.logfile))
|
|
|
|
def add_global_property(self, name: str, value: str) -> None:
|
|
__tracebackhide__ = True
|
|
_check_record_param_type("name", name)
|
|
self.global_properties.append((name, bin_xml_escape(value)))
|
|
|
|
def _get_global_properties_node(self) -> Union[py.xml.Tag, str]:
|
|
"""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 ""
|