Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1""" 

2 report test results in JUnit-XML format, 

3 for use with Jenkins and build integration servers. 

4 

5 

6Based on initial code from Ross Lawley. 

7 

8Output conforms to https://github.com/jenkinsci/xunit-plugin/blob/master/ 

9src/main/resources/org/jenkinsci/plugins/xunit/types/model/xsd/junit-10.xsd 

10""" 

11import functools 

12import os 

13import platform 

14import re 

15import sys 

16import time 

17from datetime import datetime 

18 

19import py 

20 

21import pytest 

22from _pytest import deprecated 

23from _pytest import nodes 

24from _pytest.config import filename_arg 

25from _pytest.warnings import _issue_warning_captured 

26 

27 

28class Junit(py.xml.Namespace): 

29 pass 

30 

31 

32# We need to get the subset of the invalid unicode ranges according to 

33# XML 1.0 which are valid in this python build. Hence we calculate 

34# this dynamically instead of hardcoding it. The spec range of valid 

35# chars is: Char ::= #x9 | #xA | #xD | [#x20-#xD7FF] | [#xE000-#xFFFD] 

36# | [#x10000-#x10FFFF] 

37_legal_chars = (0x09, 0x0A, 0x0D) 

38_legal_ranges = ((0x20, 0x7E), (0x80, 0xD7FF), (0xE000, 0xFFFD), (0x10000, 0x10FFFF)) 

39_legal_xml_re = [ 

40 "{}-{}".format(chr(low), chr(high)) 

41 for (low, high) in _legal_ranges 

42 if low < sys.maxunicode 

43] 

44_legal_xml_re = [chr(x) for x in _legal_chars] + _legal_xml_re 

45illegal_xml_re = re.compile("[^%s]" % "".join(_legal_xml_re)) 

46del _legal_chars 

47del _legal_ranges 

48del _legal_xml_re 

49 

50_py_ext_re = re.compile(r"\.py$") 

51 

52 

53def bin_xml_escape(arg): 

54 def repl(matchobj): 

55 i = ord(matchobj.group()) 

56 if i <= 0xFF: 

57 return "#x%02X" % i 

58 else: 

59 return "#x%04X" % i 

60 

61 return py.xml.raw(illegal_xml_re.sub(repl, py.xml.escape(arg))) 

62 

63 

64def merge_family(left, right): 

65 result = {} 

66 for kl, vl in left.items(): 

67 for kr, vr in right.items(): 

68 if not isinstance(vl, list): 

69 raise TypeError(type(vl)) 

70 result[kl] = vl + vr 

71 left.update(result) 

72 

73 

74families = {} 

75families["_base"] = {"testcase": ["classname", "name"]} 

76families["_base_legacy"] = {"testcase": ["file", "line", "url"]} 

77 

78# xUnit 1.x inherits legacy attributes 

79families["xunit1"] = families["_base"].copy() 

80merge_family(families["xunit1"], families["_base_legacy"]) 

81 

82# xUnit 2.x uses strict base attributes 

83families["xunit2"] = families["_base"] 

84 

85 

86class _NodeReporter: 

87 def __init__(self, nodeid, xml): 

88 self.id = nodeid 

89 self.xml = xml 

90 self.add_stats = self.xml.add_stats 

91 self.family = self.xml.family 

92 self.duration = 0 

93 self.properties = [] 

94 self.nodes = [] 

95 self.testcase = None 

96 self.attrs = {} 

97 

98 def append(self, node): 

99 self.xml.add_stats(type(node).__name__) 

100 self.nodes.append(node) 

101 

102 def add_property(self, name, value): 

103 self.properties.append((str(name), bin_xml_escape(value))) 

104 

105 def add_attribute(self, name, value): 

106 self.attrs[str(name)] = bin_xml_escape(value) 

107 

108 def make_properties_node(self): 

109 """Return a Junit node containing custom properties, if any. 

110 """ 

111 if self.properties: 

112 return Junit.properties( 

113 [ 

114 Junit.property(name=name, value=value) 

115 for name, value in self.properties 

116 ] 

117 ) 

118 return "" 

119 

120 def record_testreport(self, testreport): 

121 assert not self.testcase 

122 names = mangle_test_address(testreport.nodeid) 

123 existing_attrs = self.attrs 

124 classnames = names[:-1] 

125 if self.xml.prefix: 

126 classnames.insert(0, self.xml.prefix) 

127 attrs = { 

128 "classname": ".".join(classnames), 

129 "name": bin_xml_escape(names[-1]), 

130 "file": testreport.location[0], 

131 } 

132 if testreport.location[1] is not None: 

133 attrs["line"] = testreport.location[1] 

134 if hasattr(testreport, "url"): 

135 attrs["url"] = testreport.url 

136 self.attrs = attrs 

137 self.attrs.update(existing_attrs) # restore any user-defined attributes 

138 

139 # Preserve legacy testcase behavior 

140 if self.family == "xunit1": 

141 return 

142 

143 # Filter out attributes not permitted by this test family. 

144 # Including custom attributes because they are not valid here. 

145 temp_attrs = {} 

146 for key in self.attrs.keys(): 

147 if key in families[self.family]["testcase"]: 

148 temp_attrs[key] = self.attrs[key] 

149 self.attrs = temp_attrs 

150 

151 def to_xml(self): 

152 testcase = Junit.testcase(time="%.3f" % self.duration, **self.attrs) 

153 testcase.append(self.make_properties_node()) 

154 for node in self.nodes: 

155 testcase.append(node) 

156 return testcase 

157 

158 def _add_simple(self, kind, message, data=None): 

159 data = bin_xml_escape(data) 

160 node = kind(data, message=message) 

161 self.append(node) 

162 

163 def write_captured_output(self, report): 

164 if not self.xml.log_passing_tests and report.passed: 

165 return 

166 

167 content_out = report.capstdout 

168 content_log = report.caplog 

169 content_err = report.capstderr 

170 

171 if content_log or content_out: 

172 if content_log and self.xml.logging == "system-out": 

173 if content_out: 

174 # syncing stdout and the log-output is not done yet. It's 

175 # probably not worth the effort. Therefore, first the captured 

176 # stdout is shown and then the captured logs. 

177 content = "\n".join( 

178 [ 

179 " Captured Stdout ".center(80, "-"), 

180 content_out, 

181 "", 

182 " Captured Log ".center(80, "-"), 

183 content_log, 

184 ] 

185 ) 

186 else: 

187 content = content_log 

188 else: 

189 content = content_out 

190 

191 if content: 

192 tag = getattr(Junit, "system-out") 

193 self.append(tag(bin_xml_escape(content))) 

194 

195 if content_log or content_err: 

196 if content_log and self.xml.logging == "system-err": 

197 if content_err: 

198 content = "\n".join( 

199 [ 

200 " Captured Stderr ".center(80, "-"), 

201 content_err, 

202 "", 

203 " Captured Log ".center(80, "-"), 

204 content_log, 

205 ] 

206 ) 

207 else: 

208 content = content_log 

209 else: 

210 content = content_err 

211 

212 if content: 

213 tag = getattr(Junit, "system-err") 

214 self.append(tag(bin_xml_escape(content))) 

215 

216 def append_pass(self, report): 

217 self.add_stats("passed") 

218 

219 def append_failure(self, report): 

220 # msg = str(report.longrepr.reprtraceback.extraline) 

221 if hasattr(report, "wasxfail"): 

222 self._add_simple(Junit.skipped, "xfail-marked test passes unexpectedly") 

223 else: 

224 if hasattr(report.longrepr, "reprcrash"): 

225 message = report.longrepr.reprcrash.message 

226 elif isinstance(report.longrepr, str): 

227 message = report.longrepr 

228 else: 

229 message = str(report.longrepr) 

230 message = bin_xml_escape(message) 

231 fail = Junit.failure(message=message) 

232 fail.append(bin_xml_escape(report.longrepr)) 

233 self.append(fail) 

234 

235 def append_collect_error(self, report): 

236 # msg = str(report.longrepr.reprtraceback.extraline) 

237 self.append( 

238 Junit.error(bin_xml_escape(report.longrepr), message="collection failure") 

239 ) 

240 

241 def append_collect_skipped(self, report): 

242 self._add_simple(Junit.skipped, "collection skipped", report.longrepr) 

243 

244 def append_error(self, report): 

245 if report.when == "teardown": 

246 msg = "test teardown failure" 

247 else: 

248 msg = "test setup failure" 

249 self._add_simple(Junit.error, msg, report.longrepr) 

250 

251 def append_skipped(self, report): 

252 if hasattr(report, "wasxfail"): 

253 xfailreason = report.wasxfail 

254 if xfailreason.startswith("reason: "): 

255 xfailreason = xfailreason[8:] 

256 self.append( 

257 Junit.skipped( 

258 "", type="pytest.xfail", message=bin_xml_escape(xfailreason) 

259 ) 

260 ) 

261 else: 

262 filename, lineno, skipreason = report.longrepr 

263 if skipreason.startswith("Skipped: "): 

264 skipreason = skipreason[9:] 

265 details = "{}:{}: {}".format(filename, lineno, skipreason) 

266 

267 self.append( 

268 Junit.skipped( 

269 bin_xml_escape(details), 

270 type="pytest.skip", 

271 message=bin_xml_escape(skipreason), 

272 ) 

273 ) 

274 self.write_captured_output(report) 

275 

276 def finalize(self): 

277 data = self.to_xml().unicode(indent=0) 

278 self.__dict__.clear() 

279 self.to_xml = lambda: py.xml.raw(data) 

280 

281 

282def _warn_incompatibility_with_xunit2(request, fixture_name): 

283 """Emits a PytestWarning about the given fixture being incompatible with newer xunit revisions""" 

284 from _pytest.warning_types import PytestWarning 

285 

286 xml = getattr(request.config, "_xml", None) 

287 if xml is not None and xml.family not in ("xunit1", "legacy"): 

288 request.node.warn( 

289 PytestWarning( 

290 "{fixture_name} is incompatible with junit_family '{family}' (use 'legacy' or 'xunit1')".format( 

291 fixture_name=fixture_name, family=xml.family 

292 ) 

293 ) 

294 ) 

295 

296 

297@pytest.fixture 

298def record_property(request): 

299 """Add an extra properties the calling test. 

300 User properties become part of the test report and are available to the 

301 configured reporters, like JUnit XML. 

302 The fixture is callable with ``(name, value)``, with value being automatically 

303 xml-encoded. 

304 

305 Example:: 

306 

307 def test_function(record_property): 

308 record_property("example_key", 1) 

309 """ 

310 _warn_incompatibility_with_xunit2(request, "record_property") 

311 

312 def append_property(name, value): 

313 request.node.user_properties.append((name, value)) 

314 

315 return append_property 

316 

317 

318@pytest.fixture 

319def record_xml_attribute(request): 

320 """Add extra xml attributes to the tag for the calling test. 

321 The fixture is callable with ``(name, value)``, with value being 

322 automatically xml-encoded 

323 """ 

324 from _pytest.warning_types import PytestExperimentalApiWarning 

325 

326 request.node.warn( 

327 PytestExperimentalApiWarning("record_xml_attribute is an experimental feature") 

328 ) 

329 

330 _warn_incompatibility_with_xunit2(request, "record_xml_attribute") 

331 

332 # Declare noop 

333 def add_attr_noop(name, value): 

334 pass 

335 

336 attr_func = add_attr_noop 

337 

338 xml = getattr(request.config, "_xml", None) 

339 if xml is not None: 

340 node_reporter = xml.node_reporter(request.node.nodeid) 

341 attr_func = node_reporter.add_attribute 

342 

343 return attr_func 

344 

345 

346def _check_record_param_type(param, v): 

347 """Used by record_testsuite_property to check that the given parameter name is of the proper 

348 type""" 

349 __tracebackhide__ = True 

350 if not isinstance(v, str): 

351 msg = "{param} parameter needs to be a string, but {g} given" 

352 raise TypeError(msg.format(param=param, g=type(v).__name__)) 

353 

354 

355@pytest.fixture(scope="session") 

356def record_testsuite_property(request): 

357 """ 

358 Records a new ``<property>`` tag as child of the root ``<testsuite>``. This is suitable to 

359 writing global information regarding the entire test suite, and is compatible with ``xunit2`` JUnit family. 

360 

361 This is a ``session``-scoped fixture which is called with ``(name, value)``. Example: 

362 

363 .. code-block:: python 

364 

365 def test_foo(record_testsuite_property): 

366 record_testsuite_property("ARCH", "PPC") 

367 record_testsuite_property("STORAGE_TYPE", "CEPH") 

368 

369 ``name`` must be a string, ``value`` will be converted to a string and properly xml-escaped. 

370 """ 

371 

372 __tracebackhide__ = True 

373 

374 def record_func(name, value): 

375 """noop function in case --junitxml was not passed in the command-line""" 

376 __tracebackhide__ = True 

377 _check_record_param_type("name", name) 

378 

379 xml = getattr(request.config, "_xml", None) 

380 if xml is not None: 

381 record_func = xml.add_global_property # noqa 

382 return record_func 

383 

384 

385def pytest_addoption(parser): 

386 group = parser.getgroup("terminal reporting") 

387 group.addoption( 

388 "--junitxml", 

389 "--junit-xml", 

390 action="store", 

391 dest="xmlpath", 

392 metavar="path", 

393 type=functools.partial(filename_arg, optname="--junitxml"), 

394 default=None, 

395 help="create junit-xml style report file at given path.", 

396 ) 

397 group.addoption( 

398 "--junitprefix", 

399 "--junit-prefix", 

400 action="store", 

401 metavar="str", 

402 default=None, 

403 help="prepend prefix to classnames in junit-xml output", 

404 ) 

405 parser.addini( 

406 "junit_suite_name", "Test suite name for JUnit report", default="pytest" 

407 ) 

408 parser.addini( 

409 "junit_logging", 

410 "Write captured log messages to JUnit report: " 

411 "one of no|system-out|system-err", 

412 default="no", 

413 ) # choices=['no', 'stdout', 'stderr']) 

414 parser.addini( 

415 "junit_log_passing_tests", 

416 "Capture log information for passing tests to JUnit report: ", 

417 type="bool", 

418 default=True, 

419 ) 

420 parser.addini( 

421 "junit_duration_report", 

422 "Duration time to report: one of total|call", 

423 default="total", 

424 ) # choices=['total', 'call']) 

425 parser.addini( 

426 "junit_family", "Emit XML for schema: one of legacy|xunit1|xunit2", default=None 

427 ) 

428 

429 

430def pytest_configure(config): 

431 xmlpath = config.option.xmlpath 

432 # prevent opening xmllog on slave nodes (xdist) 

433 if xmlpath and not hasattr(config, "slaveinput"): 

434 junit_family = config.getini("junit_family") 

435 if not junit_family: 

436 _issue_warning_captured(deprecated.JUNIT_XML_DEFAULT_FAMILY, config.hook, 2) 

437 junit_family = "xunit1" 

438 config._xml = LogXML( 

439 xmlpath, 

440 config.option.junitprefix, 

441 config.getini("junit_suite_name"), 

442 config.getini("junit_logging"), 

443 config.getini("junit_duration_report"), 

444 junit_family, 

445 config.getini("junit_log_passing_tests"), 

446 ) 

447 config.pluginmanager.register(config._xml) 

448 

449 

450def pytest_unconfigure(config): 

451 xml = getattr(config, "_xml", None) 

452 if xml: 

453 del config._xml 

454 config.pluginmanager.unregister(xml) 

455 

456 

457def mangle_test_address(address): 

458 path, possible_open_bracket, params = address.partition("[") 

459 names = path.split("::") 

460 try: 

461 names.remove("()") 

462 except ValueError: 

463 pass 

464 # convert file path to dotted path 

465 names[0] = names[0].replace(nodes.SEP, ".") 

466 names[0] = _py_ext_re.sub("", names[0]) 

467 # put any params back 

468 names[-1] += possible_open_bracket + params 

469 return names 

470 

471 

472class LogXML: 

473 def __init__( 

474 self, 

475 logfile, 

476 prefix, 

477 suite_name="pytest", 

478 logging="no", 

479 report_duration="total", 

480 family="xunit1", 

481 log_passing_tests=True, 

482 ): 

483 logfile = os.path.expanduser(os.path.expandvars(logfile)) 

484 self.logfile = os.path.normpath(os.path.abspath(logfile)) 

485 self.prefix = prefix 

486 self.suite_name = suite_name 

487 self.logging = logging 

488 self.log_passing_tests = log_passing_tests 

489 self.report_duration = report_duration 

490 self.family = family 

491 self.stats = dict.fromkeys(["error", "passed", "failure", "skipped"], 0) 

492 self.node_reporters = {} # nodeid -> _NodeReporter 

493 self.node_reporters_ordered = [] 

494 self.global_properties = [] 

495 

496 # List of reports that failed on call but teardown is pending. 

497 self.open_reports = [] 

498 self.cnt_double_fail_tests = 0 

499 

500 # Replaces convenience family with real family 

501 if self.family == "legacy": 

502 self.family = "xunit1" 

503 

504 def finalize(self, report): 

505 nodeid = getattr(report, "nodeid", report) 

506 # local hack to handle xdist report order 

507 slavenode = getattr(report, "node", None) 

508 reporter = self.node_reporters.pop((nodeid, slavenode)) 

509 if reporter is not None: 

510 reporter.finalize() 

511 

512 def node_reporter(self, report): 

513 nodeid = getattr(report, "nodeid", report) 

514 # local hack to handle xdist report order 

515 slavenode = getattr(report, "node", None) 

516 

517 key = nodeid, slavenode 

518 

519 if key in self.node_reporters: 

520 # TODO: breaks for --dist=each 

521 return self.node_reporters[key] 

522 

523 reporter = _NodeReporter(nodeid, self) 

524 

525 self.node_reporters[key] = reporter 

526 self.node_reporters_ordered.append(reporter) 

527 

528 return reporter 

529 

530 def add_stats(self, key): 

531 if key in self.stats: 

532 self.stats[key] += 1 

533 

534 def _opentestcase(self, report): 

535 reporter = self.node_reporter(report) 

536 reporter.record_testreport(report) 

537 return reporter 

538 

539 def pytest_runtest_logreport(self, report): 

540 """handle a setup/call/teardown report, generating the appropriate 

541 xml tags as necessary. 

542 

543 note: due to plugins like xdist, this hook may be called in interlaced 

544 order with reports from other nodes. for example: 

545 

546 usual call order: 

547 -> setup node1 

548 -> call node1 

549 -> teardown node1 

550 -> setup node2 

551 -> call node2 

552 -> teardown node2 

553 

554 possible call order in xdist: 

555 -> setup node1 

556 -> call node1 

557 -> setup node2 

558 -> call node2 

559 -> teardown node2 

560 -> teardown node1 

561 """ 

562 close_report = None 

563 if report.passed: 

564 if report.when == "call": # ignore setup/teardown 

565 reporter = self._opentestcase(report) 

566 reporter.append_pass(report) 

567 elif report.failed: 

568 if report.when == "teardown": 

569 # The following vars are needed when xdist plugin is used 

570 report_wid = getattr(report, "worker_id", None) 

571 report_ii = getattr(report, "item_index", None) 

572 close_report = next( 

573 ( 

574 rep 

575 for rep in self.open_reports 

576 if ( 

577 rep.nodeid == report.nodeid 

578 and getattr(rep, "item_index", None) == report_ii 

579 and getattr(rep, "worker_id", None) == report_wid 

580 ) 

581 ), 

582 None, 

583 ) 

584 if close_report: 

585 # We need to open new testcase in case we have failure in 

586 # call and error in teardown in order to follow junit 

587 # schema 

588 self.finalize(close_report) 

589 self.cnt_double_fail_tests += 1 

590 reporter = self._opentestcase(report) 

591 if report.when == "call": 

592 reporter.append_failure(report) 

593 self.open_reports.append(report) 

594 if not self.log_passing_tests: 

595 reporter.write_captured_output(report) 

596 else: 

597 reporter.append_error(report) 

598 elif report.skipped: 

599 reporter = self._opentestcase(report) 

600 reporter.append_skipped(report) 

601 self.update_testcase_duration(report) 

602 if report.when == "teardown": 

603 reporter = self._opentestcase(report) 

604 reporter.write_captured_output(report) 

605 

606 for propname, propvalue in report.user_properties: 

607 reporter.add_property(propname, propvalue) 

608 

609 self.finalize(report) 

610 report_wid = getattr(report, "worker_id", None) 

611 report_ii = getattr(report, "item_index", None) 

612 close_report = next( 

613 ( 

614 rep 

615 for rep in self.open_reports 

616 if ( 

617 rep.nodeid == report.nodeid 

618 and getattr(rep, "item_index", None) == report_ii 

619 and getattr(rep, "worker_id", None) == report_wid 

620 ) 

621 ), 

622 None, 

623 ) 

624 if close_report: 

625 self.open_reports.remove(close_report) 

626 

627 def update_testcase_duration(self, report): 

628 """accumulates total duration for nodeid from given report and updates 

629 the Junit.testcase with the new total if already created. 

630 """ 

631 if self.report_duration == "total" or report.when == self.report_duration: 

632 reporter = self.node_reporter(report) 

633 reporter.duration += getattr(report, "duration", 0.0) 

634 

635 def pytest_collectreport(self, report): 

636 if not report.passed: 

637 reporter = self._opentestcase(report) 

638 if report.failed: 

639 reporter.append_collect_error(report) 

640 else: 

641 reporter.append_collect_skipped(report) 

642 

643 def pytest_internalerror(self, excrepr): 

644 reporter = self.node_reporter("internal") 

645 reporter.attrs.update(classname="pytest", name="internal") 

646 reporter._add_simple(Junit.error, "internal error", excrepr) 

647 

648 def pytest_sessionstart(self): 

649 self.suite_start_time = time.time() 

650 

651 def pytest_sessionfinish(self): 

652 dirname = os.path.dirname(os.path.abspath(self.logfile)) 

653 if not os.path.isdir(dirname): 

654 os.makedirs(dirname) 

655 logfile = open(self.logfile, "w", encoding="utf-8") 

656 suite_stop_time = time.time() 

657 suite_time_delta = suite_stop_time - self.suite_start_time 

658 

659 numtests = ( 

660 self.stats["passed"] 

661 + self.stats["failure"] 

662 + self.stats["skipped"] 

663 + self.stats["error"] 

664 - self.cnt_double_fail_tests 

665 ) 

666 logfile.write('<?xml version="1.0" encoding="utf-8"?>') 

667 

668 suite_node = Junit.testsuite( 

669 self._get_global_properties_node(), 

670 [x.to_xml() for x in self.node_reporters_ordered], 

671 name=self.suite_name, 

672 errors=self.stats["error"], 

673 failures=self.stats["failure"], 

674 skipped=self.stats["skipped"], 

675 tests=numtests, 

676 time="%.3f" % suite_time_delta, 

677 timestamp=datetime.fromtimestamp(self.suite_start_time).isoformat(), 

678 hostname=platform.node(), 

679 ) 

680 logfile.write(Junit.testsuites([suite_node]).unicode(indent=0)) 

681 logfile.close() 

682 

683 def pytest_terminal_summary(self, terminalreporter): 

684 terminalreporter.write_sep("-", "generated xml file: %s" % (self.logfile)) 

685 

686 def add_global_property(self, name, value): 

687 __tracebackhide__ = True 

688 _check_record_param_type("name", name) 

689 self.global_properties.append((name, bin_xml_escape(value))) 

690 

691 def _get_global_properties_node(self): 

692 """Return a Junit node containing custom properties, if any. 

693 """ 

694 if self.global_properties: 

695 return Junit.properties( 

696 [ 

697 Junit.property(name=name, value=value) 

698 for name, value in self.global_properties 

699 ] 

700 ) 

701 return ""