terminal.py 63 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763
  1. # mypy: allow-untyped-defs
  2. """Terminal reporting of the full testing process.
  3. This is a good source for looking at the various reporting hooks.
  4. """
  5. from __future__ import annotations
  6. import argparse
  7. from collections import Counter
  8. from collections.abc import Callable
  9. from collections.abc import Generator
  10. from collections.abc import Mapping
  11. from collections.abc import Sequence
  12. import dataclasses
  13. import datetime
  14. from functools import partial
  15. import inspect
  16. from pathlib import Path
  17. import platform
  18. import sys
  19. import textwrap
  20. from typing import Any
  21. from typing import ClassVar
  22. from typing import final
  23. from typing import Literal
  24. from typing import NamedTuple
  25. from typing import TextIO
  26. from typing import TYPE_CHECKING
  27. import warnings
  28. import pluggy
  29. from _pytest import compat
  30. from _pytest import nodes
  31. from _pytest import timing
  32. from _pytest._code import ExceptionInfo
  33. from _pytest._code.code import ExceptionRepr
  34. from _pytest._io import TerminalWriter
  35. from _pytest._io.wcwidth import wcswidth
  36. import _pytest._version
  37. from _pytest.compat import running_on_ci
  38. from _pytest.config import _PluggyPlugin
  39. from _pytest.config import Config
  40. from _pytest.config import ExitCode
  41. from _pytest.config import hookimpl
  42. from _pytest.config.argparsing import Parser
  43. from _pytest.nodes import Item
  44. from _pytest.nodes import Node
  45. from _pytest.pathlib import absolutepath
  46. from _pytest.pathlib import bestrelpath
  47. from _pytest.reports import BaseReport
  48. from _pytest.reports import CollectReport
  49. from _pytest.reports import TestReport
  50. if TYPE_CHECKING:
  51. from _pytest.main import Session
  52. REPORT_COLLECTING_RESOLUTION = 0.5
  53. KNOWN_TYPES = (
  54. "failed",
  55. "passed",
  56. "skipped",
  57. "deselected",
  58. "xfailed",
  59. "xpassed",
  60. "warnings",
  61. "error",
  62. "subtests passed",
  63. "subtests failed",
  64. "subtests skipped",
  65. )
  66. _REPORTCHARS_DEFAULT = "fE"
  67. class MoreQuietAction(argparse.Action):
  68. """A modified copy of the argparse count action which counts down and updates
  69. the legacy quiet attribute at the same time.
  70. Used to unify verbosity handling.
  71. """
  72. def __init__(
  73. self,
  74. option_strings: Sequence[str],
  75. dest: str,
  76. default: object = None,
  77. required: bool = False,
  78. help: str | None = None,
  79. ) -> None:
  80. super().__init__(
  81. option_strings=option_strings,
  82. dest=dest,
  83. nargs=0,
  84. default=default,
  85. required=required,
  86. help=help,
  87. )
  88. def __call__(
  89. self,
  90. parser: argparse.ArgumentParser,
  91. namespace: argparse.Namespace,
  92. values: str | Sequence[object] | None,
  93. option_string: str | None = None,
  94. ) -> None:
  95. new_count = getattr(namespace, self.dest, 0) - 1
  96. setattr(namespace, self.dest, new_count)
  97. # todo Deprecate config.quiet
  98. namespace.quiet = getattr(namespace, "quiet", 0) + 1
  99. class TestShortLogReport(NamedTuple):
  100. """Used to store the test status result category, shortletter and verbose word.
  101. For example ``"rerun", "R", ("RERUN", {"yellow": True})``.
  102. :ivar category:
  103. The class of result, for example ``“passed”``, ``“skipped”``, ``“error”``, or the empty string.
  104. :ivar letter:
  105. The short letter shown as testing progresses, for example ``"."``, ``"s"``, ``"E"``, or the empty string.
  106. :ivar word:
  107. Verbose word is shown as testing progresses in verbose mode, for example ``"PASSED"``, ``"SKIPPED"``,
  108. ``"ERROR"``, or the empty string.
  109. """
  110. category: str
  111. letter: str
  112. word: str | tuple[str, Mapping[str, bool]]
  113. def pytest_addoption(parser: Parser) -> None:
  114. group = parser.getgroup("terminal reporting", "Reporting", after="general")
  115. group._addoption( # private to use reserved lower-case short option
  116. "-v",
  117. "--verbose",
  118. action="count",
  119. default=0,
  120. dest="verbose",
  121. help="Increase verbosity",
  122. )
  123. group.addoption(
  124. "--no-header",
  125. action="store_true",
  126. default=False,
  127. dest="no_header",
  128. help="Disable header",
  129. )
  130. group.addoption(
  131. "--no-summary",
  132. action="store_true",
  133. default=False,
  134. dest="no_summary",
  135. help="Disable summary",
  136. )
  137. group.addoption(
  138. "--no-fold-skipped",
  139. action="store_false",
  140. dest="fold_skipped",
  141. default=True,
  142. help="Do not fold skipped tests in short summary.",
  143. )
  144. group.addoption(
  145. "--force-short-summary",
  146. action="store_true",
  147. dest="force_short_summary",
  148. default=False,
  149. help="Force condensed summary output regardless of verbosity level.",
  150. )
  151. group._addoption( # private to use reserved lower-case short option
  152. "-q",
  153. "--quiet",
  154. action=MoreQuietAction,
  155. default=0,
  156. dest="verbose",
  157. help="Decrease verbosity",
  158. )
  159. group.addoption(
  160. "--verbosity",
  161. dest="verbose",
  162. type=int,
  163. default=0,
  164. help="Set verbosity. Default: 0.",
  165. )
  166. group._addoption( # private to use reserved lower-case short option
  167. "-r",
  168. action="store",
  169. dest="reportchars",
  170. default=_REPORTCHARS_DEFAULT,
  171. metavar="chars",
  172. help="Show extra test summary info as specified by chars: (f)ailed, "
  173. "(E)rror, (s)kipped, (x)failed, (X)passed, "
  174. "(p)assed, (P)assed with output, (a)ll except passed (p/P), or (A)ll. "
  175. "(w)arnings are enabled by default (see --disable-warnings), "
  176. "'N' can be used to reset the list. (default: 'fE').",
  177. )
  178. group.addoption(
  179. "--disable-warnings",
  180. "--disable-pytest-warnings",
  181. default=False,
  182. dest="disable_warnings",
  183. action="store_true",
  184. help="Disable warnings summary",
  185. )
  186. group._addoption( # private to use reserved lower-case short option
  187. "-l",
  188. "--showlocals",
  189. action="store_true",
  190. dest="showlocals",
  191. default=False,
  192. help="Show locals in tracebacks (disabled by default)",
  193. )
  194. group.addoption(
  195. "--no-showlocals",
  196. action="store_false",
  197. dest="showlocals",
  198. help="Hide locals in tracebacks (negate --showlocals passed through addopts)",
  199. )
  200. group.addoption(
  201. "--tb",
  202. metavar="style",
  203. action="store",
  204. dest="tbstyle",
  205. default="auto",
  206. choices=["auto", "long", "short", "no", "line", "native"],
  207. help="Traceback print mode (auto/long/short/line/native/no)",
  208. )
  209. group.addoption(
  210. "--xfail-tb",
  211. action="store_true",
  212. dest="xfail_tb",
  213. default=False,
  214. help="Show tracebacks for xfail (as long as --tb != no)",
  215. )
  216. group.addoption(
  217. "--show-capture",
  218. action="store",
  219. dest="showcapture",
  220. choices=["no", "stdout", "stderr", "log", "all"],
  221. default="all",
  222. help="Controls how captured stdout/stderr/log is shown on failed tests. "
  223. "Default: all.",
  224. )
  225. group.addoption(
  226. "--fulltrace",
  227. "--full-trace",
  228. action="store_true",
  229. default=False,
  230. help="Don't cut any tracebacks (default is to cut)",
  231. )
  232. group.addoption(
  233. "--color",
  234. metavar="color",
  235. action="store",
  236. dest="color",
  237. default="auto",
  238. choices=["yes", "no", "auto"],
  239. help="Color terminal output (yes/no/auto)",
  240. )
  241. group.addoption(
  242. "--code-highlight",
  243. default="yes",
  244. choices=["yes", "no"],
  245. help="Whether code should be highlighted (only if --color is also enabled). "
  246. "Default: yes.",
  247. )
  248. parser.addini(
  249. "console_output_style",
  250. help='Console output: "classic", or with additional progress information '
  251. '("progress" (percentage) | "count" | "progress-even-when-capture-no" (forces '
  252. "progress even when capture=no)",
  253. default="progress",
  254. )
  255. Config._add_verbosity_ini(
  256. parser,
  257. Config.VERBOSITY_TEST_CASES,
  258. help=(
  259. "Specify a verbosity level for test case execution, overriding the main level. "
  260. "Higher levels will provide more detailed information about each test case executed."
  261. ),
  262. )
  263. def pytest_configure(config: Config) -> None:
  264. reporter = TerminalReporter(config, sys.stdout)
  265. config.pluginmanager.register(reporter, "terminalreporter")
  266. if config.option.debug or config.option.traceconfig:
  267. def mywriter(tags, args):
  268. msg = " ".join(map(str, args))
  269. reporter.write_line("[traceconfig] " + msg)
  270. config.trace.root.setprocessor("pytest:config", mywriter)
  271. # See terminalprogress.py.
  272. # On Windows it's safe to load by default.
  273. if sys.platform == "win32":
  274. config.pluginmanager.import_plugin("terminalprogress")
  275. def getreportopt(config: Config) -> str:
  276. reportchars: str = config.option.reportchars
  277. old_aliases = {"F", "S"}
  278. reportopts = ""
  279. for char in reportchars:
  280. if char in old_aliases:
  281. char = char.lower()
  282. if char == "a":
  283. reportopts = "sxXEf"
  284. elif char == "A":
  285. reportopts = "PpsxXEf"
  286. elif char == "N":
  287. reportopts = ""
  288. elif char not in reportopts:
  289. reportopts += char
  290. if not config.option.disable_warnings and "w" not in reportopts:
  291. reportopts = "w" + reportopts
  292. elif config.option.disable_warnings and "w" in reportopts:
  293. reportopts = reportopts.replace("w", "")
  294. return reportopts
  295. @hookimpl(trylast=True) # after _pytest.runner
  296. def pytest_report_teststatus(report: BaseReport) -> tuple[str, str, str]:
  297. letter = "F"
  298. if report.passed:
  299. letter = "."
  300. elif report.skipped:
  301. letter = "s"
  302. outcome: str = report.outcome
  303. if report.when in ("collect", "setup", "teardown") and outcome == "failed":
  304. outcome = "error"
  305. letter = "E"
  306. return outcome, letter, outcome.upper()
  307. @dataclasses.dataclass
  308. class WarningReport:
  309. """Simple structure to hold warnings information captured by ``pytest_warning_recorded``.
  310. :ivar str message:
  311. User friendly message about the warning.
  312. :ivar str|None nodeid:
  313. nodeid that generated the warning (see ``get_location``).
  314. :ivar tuple fslocation:
  315. File system location of the source of the warning (see ``get_location``).
  316. """
  317. message: str
  318. nodeid: str | None = None
  319. fslocation: tuple[str, int] | None = None
  320. count_towards_summary: ClassVar = True
  321. def get_location(self, config: Config) -> str | None:
  322. """Return the more user-friendly information about the location of a warning, or None."""
  323. if self.nodeid:
  324. return self.nodeid
  325. if self.fslocation:
  326. filename, linenum = self.fslocation
  327. relpath = bestrelpath(config.invocation_params.dir, absolutepath(filename))
  328. return f"{relpath}:{linenum}"
  329. return None
  330. @final
  331. class TerminalReporter:
  332. def __init__(self, config: Config, file: TextIO | None = None) -> None:
  333. import _pytest.config
  334. self.config = config
  335. self._numcollected = 0
  336. self._session: Session | None = None
  337. self._showfspath: bool | None = None
  338. self.stats: dict[str, list[Any]] = {}
  339. self._main_color: str | None = None
  340. self._known_types: list[str] | None = None
  341. self.startpath = config.invocation_params.dir
  342. if file is None:
  343. file = sys.stdout
  344. self._tw = _pytest.config.create_terminal_writer(config, file)
  345. self._screen_width = self._tw.fullwidth
  346. self.currentfspath: None | Path | str | int = None
  347. self.reportchars = getreportopt(config)
  348. self.foldskipped = config.option.fold_skipped
  349. self.hasmarkup = self._tw.hasmarkup
  350. # isatty should be a method but was wrongly implemented as a boolean.
  351. # We use CallableBool here to support both.
  352. self.isatty = compat.CallableBool(file.isatty())
  353. self._progress_nodeids_reported: set[str] = set()
  354. self._timing_nodeids_reported: set[str] = set()
  355. self._show_progress_info = self._determine_show_progress_info()
  356. self._collect_report_last_write = timing.Instant()
  357. self._already_displayed_warnings: int | None = None
  358. self._keyboardinterrupt_memo: ExceptionRepr | None = None
  359. def _determine_show_progress_info(
  360. self,
  361. ) -> Literal["progress", "count", "times", False]:
  362. """Return whether we should display progress information based on the current config."""
  363. # do not show progress if we are not capturing output (#3038) unless explicitly
  364. # overridden by progress-even-when-capture-no
  365. if (
  366. self.config.getoption("capture", "no") == "no"
  367. and self.config.getini("console_output_style")
  368. != "progress-even-when-capture-no"
  369. ):
  370. return False
  371. # do not show progress if we are showing fixture setup/teardown
  372. if self.config.getoption("setupshow", False):
  373. return False
  374. cfg: str = self.config.getini("console_output_style")
  375. if cfg in {"progress", "progress-even-when-capture-no"}:
  376. return "progress"
  377. elif cfg == "count":
  378. return "count"
  379. elif cfg == "times":
  380. return "times"
  381. else:
  382. return False
  383. @property
  384. def verbosity(self) -> int:
  385. verbosity: int = self.config.option.verbose
  386. return verbosity
  387. @property
  388. def showheader(self) -> bool:
  389. return self.verbosity >= 0
  390. @property
  391. def no_header(self) -> bool:
  392. return bool(self.config.option.no_header)
  393. @property
  394. def no_summary(self) -> bool:
  395. return bool(self.config.option.no_summary)
  396. @property
  397. def showfspath(self) -> bool:
  398. if self._showfspath is None:
  399. return self.config.get_verbosity(Config.VERBOSITY_TEST_CASES) >= 0
  400. return self._showfspath
  401. @showfspath.setter
  402. def showfspath(self, value: bool | None) -> None:
  403. self._showfspath = value
  404. @property
  405. def showlongtestinfo(self) -> bool:
  406. return self.config.get_verbosity(Config.VERBOSITY_TEST_CASES) > 0
  407. @property
  408. def reported_progress(self) -> int:
  409. """The amount of items reported in the progress so far.
  410. :meta private:
  411. """
  412. return len(self._progress_nodeids_reported)
  413. def hasopt(self, char: str) -> bool:
  414. char = {"xfailed": "x", "skipped": "s"}.get(char, char)
  415. return char in self.reportchars
  416. def write_fspath_result(self, nodeid: str, res: str, **markup: bool) -> None:
  417. fspath = self.config.rootpath / nodeid.split("::")[0]
  418. if self.currentfspath is None or fspath != self.currentfspath:
  419. if self.currentfspath is not None and self._show_progress_info:
  420. self._write_progress_information_filling_space()
  421. self.currentfspath = fspath
  422. relfspath = bestrelpath(self.startpath, fspath)
  423. self._tw.line()
  424. self._tw.write(relfspath + " ")
  425. self._tw.write(res, flush=True, **markup)
  426. def write_ensure_prefix(self, prefix: str, extra: str = "", **kwargs) -> None:
  427. if self.currentfspath != prefix:
  428. self._tw.line()
  429. self.currentfspath = prefix
  430. self._tw.write(prefix)
  431. if extra:
  432. self._tw.write(extra, **kwargs)
  433. self.currentfspath = -2
  434. def ensure_newline(self) -> None:
  435. if self.currentfspath:
  436. self._tw.line()
  437. self.currentfspath = None
  438. def wrap_write(
  439. self,
  440. content: str,
  441. *,
  442. flush: bool = False,
  443. margin: int = 8,
  444. line_sep: str = "\n",
  445. **markup: bool,
  446. ) -> None:
  447. """Wrap message with margin for progress info."""
  448. width_of_current_line = self._tw.width_of_current_line
  449. wrapped = line_sep.join(
  450. textwrap.wrap(
  451. " " * width_of_current_line + content,
  452. width=self._screen_width - margin,
  453. drop_whitespace=True,
  454. replace_whitespace=False,
  455. ),
  456. )
  457. wrapped = wrapped[width_of_current_line:]
  458. self._tw.write(wrapped, flush=flush, **markup)
  459. def write(self, content: str, *, flush: bool = False, **markup: bool) -> None:
  460. self._tw.write(content, flush=flush, **markup)
  461. def write_raw(self, content: str, *, flush: bool = False) -> None:
  462. self._tw.write_raw(content, flush=flush)
  463. def flush(self) -> None:
  464. self._tw.flush()
  465. def write_line(self, line: str | bytes, **markup: bool) -> None:
  466. if not isinstance(line, str):
  467. line = str(line, errors="replace")
  468. self.ensure_newline()
  469. self._tw.line(line, **markup)
  470. def rewrite(self, line: str, **markup: bool) -> None:
  471. """Rewinds the terminal cursor to the beginning and writes the given line.
  472. :param erase:
  473. If True, will also add spaces until the full terminal width to ensure
  474. previous lines are properly erased.
  475. The rest of the keyword arguments are markup instructions.
  476. """
  477. erase = markup.pop("erase", False)
  478. if erase:
  479. fill_count = self._tw.fullwidth - len(line) - 1
  480. fill = " " * fill_count
  481. else:
  482. fill = ""
  483. line = str(line)
  484. self._tw.write("\r" + line + fill, **markup)
  485. def write_sep(
  486. self,
  487. sep: str,
  488. title: str | None = None,
  489. fullwidth: int | None = None,
  490. **markup: bool,
  491. ) -> None:
  492. self.ensure_newline()
  493. self._tw.sep(sep, title, fullwidth, **markup)
  494. def section(self, title: str, sep: str = "=", **kw: bool) -> None:
  495. self._tw.sep(sep, title, **kw)
  496. def line(self, msg: str, **kw: bool) -> None:
  497. self._tw.line(msg, **kw)
  498. def _add_stats(self, category: str, items: Sequence[Any]) -> None:
  499. set_main_color = category not in self.stats
  500. self.stats.setdefault(category, []).extend(items)
  501. if set_main_color:
  502. self._set_main_color()
  503. def pytest_internalerror(self, excrepr: ExceptionRepr) -> bool:
  504. for line in str(excrepr).split("\n"):
  505. self.write_line("INTERNALERROR> " + line)
  506. return True
  507. def pytest_warning_recorded(
  508. self,
  509. warning_message: warnings.WarningMessage,
  510. nodeid: str,
  511. ) -> None:
  512. from _pytest.warnings import warning_record_to_str
  513. fslocation = warning_message.filename, warning_message.lineno
  514. message = warning_record_to_str(warning_message)
  515. warning_report = WarningReport(
  516. fslocation=fslocation, message=message, nodeid=nodeid
  517. )
  518. self._add_stats("warnings", [warning_report])
  519. def pytest_plugin_registered(self, plugin: _PluggyPlugin) -> None:
  520. if self.config.option.traceconfig:
  521. msg = f"PLUGIN registered: {plugin}"
  522. # XXX This event may happen during setup/teardown time
  523. # which unfortunately captures our output here
  524. # which garbles our output if we use self.write_line.
  525. self.write_line(msg)
  526. def pytest_deselected(self, items: Sequence[Item]) -> None:
  527. self._add_stats("deselected", items)
  528. def pytest_runtest_logstart(
  529. self, nodeid: str, location: tuple[str, int | None, str]
  530. ) -> None:
  531. fspath, lineno, domain = location
  532. # Ensure that the path is printed before the
  533. # 1st test of a module starts running.
  534. if self.showlongtestinfo:
  535. line = self._locationline(nodeid, fspath, lineno, domain)
  536. self.write_ensure_prefix(line, "")
  537. self.flush()
  538. elif self.showfspath:
  539. self.write_fspath_result(nodeid, "")
  540. self.flush()
  541. def pytest_runtest_logreport(self, report: TestReport) -> None:
  542. self._tests_ran = True
  543. rep = report
  544. res = TestShortLogReport(
  545. *self.config.hook.pytest_report_teststatus(report=rep, config=self.config)
  546. )
  547. category, letter, word = res.category, res.letter, res.word
  548. if not isinstance(word, tuple):
  549. markup = None
  550. else:
  551. word, markup = word
  552. self._add_stats(category, [rep])
  553. if not letter and not word:
  554. # Probably passed setup/teardown.
  555. return
  556. if markup is None:
  557. was_xfail = hasattr(report, "wasxfail")
  558. if rep.passed and not was_xfail:
  559. markup = {"green": True}
  560. elif rep.passed and was_xfail:
  561. markup = {"yellow": True}
  562. elif rep.failed:
  563. markup = {"red": True}
  564. elif rep.skipped:
  565. markup = {"yellow": True}
  566. else:
  567. markup = {}
  568. self._progress_nodeids_reported.add(rep.nodeid)
  569. if self.config.get_verbosity(Config.VERBOSITY_TEST_CASES) <= 0:
  570. self._tw.write(letter, **markup)
  571. # When running in xdist, the logreport and logfinish of multiple
  572. # items are interspersed, e.g. `logreport`, `logreport`,
  573. # `logfinish`, `logfinish`. To avoid the "past edge" calculation
  574. # from getting confused and overflowing (#7166), do the past edge
  575. # printing here and not in logfinish, except for the 100% which
  576. # should only be printed after all teardowns are finished.
  577. if self._show_progress_info and not self._is_last_item:
  578. self._write_progress_information_if_past_edge()
  579. else:
  580. line = self._locationline(rep.nodeid, *rep.location)
  581. running_xdist = hasattr(rep, "node")
  582. if not running_xdist:
  583. self.write_ensure_prefix(line, word, **markup)
  584. if rep.skipped or hasattr(report, "wasxfail"):
  585. reason = _get_raw_skip_reason(rep)
  586. if self.config.get_verbosity(Config.VERBOSITY_TEST_CASES) < 2:
  587. available_width = (
  588. (self._tw.fullwidth - self._tw.width_of_current_line)
  589. - len(" [100%]")
  590. - 1
  591. )
  592. formatted_reason = _format_trimmed(
  593. " ({})", reason, available_width
  594. )
  595. else:
  596. formatted_reason = f" ({reason})"
  597. if reason and formatted_reason is not None:
  598. self.wrap_write(formatted_reason)
  599. if self._show_progress_info:
  600. self._write_progress_information_filling_space()
  601. else:
  602. self.ensure_newline()
  603. self._tw.write(f"[{rep.node.gateway.id}]")
  604. if self._show_progress_info:
  605. self._tw.write(
  606. self._get_progress_information_message() + " ", cyan=True
  607. )
  608. else:
  609. self._tw.write(" ")
  610. self._tw.write(word, **markup)
  611. self._tw.write(" " + line)
  612. self.currentfspath = -2
  613. self.flush()
  614. @property
  615. def _is_last_item(self) -> bool:
  616. assert self._session is not None
  617. return self.reported_progress == self._session.testscollected
  618. @hookimpl(wrapper=True)
  619. def pytest_runtestloop(self) -> Generator[None, object, object]:
  620. result = yield
  621. # Write the final/100% progress -- deferred until the loop is complete.
  622. if (
  623. self.config.get_verbosity(Config.VERBOSITY_TEST_CASES) <= 0
  624. and self._show_progress_info
  625. and self.reported_progress
  626. ):
  627. self._write_progress_information_filling_space()
  628. return result
  629. def _get_progress_information_message(self) -> str:
  630. assert self._session
  631. collected = self._session.testscollected
  632. if self._show_progress_info == "count":
  633. if collected:
  634. progress = self.reported_progress
  635. counter_format = f"{{:{len(str(collected))}d}}"
  636. format_string = f" [{counter_format}/{{}}]"
  637. return format_string.format(progress, collected)
  638. return f" [ {collected} / {collected} ]"
  639. if self._show_progress_info == "times":
  640. if not collected:
  641. return ""
  642. all_reports = (
  643. self._get_reports_to_display("passed")
  644. + self._get_reports_to_display("xpassed")
  645. + self._get_reports_to_display("failed")
  646. + self._get_reports_to_display("xfailed")
  647. + self._get_reports_to_display("skipped")
  648. + self._get_reports_to_display("error")
  649. + self._get_reports_to_display("")
  650. )
  651. current_location = all_reports[-1].location[0]
  652. not_reported = [
  653. r for r in all_reports if r.nodeid not in self._timing_nodeids_reported
  654. ]
  655. tests_in_module = sum(
  656. i.location[0] == current_location for i in self._session.items
  657. )
  658. tests_completed = sum(
  659. r.when == "setup"
  660. for r in not_reported
  661. if r.location[0] == current_location
  662. )
  663. last_in_module = tests_completed == tests_in_module
  664. if self.showlongtestinfo or last_in_module:
  665. self._timing_nodeids_reported.update(r.nodeid for r in not_reported)
  666. return format_node_duration(
  667. sum(r.duration for r in not_reported if isinstance(r, TestReport))
  668. )
  669. return ""
  670. if collected:
  671. return f" [{self.reported_progress * 100 // collected:3d}%]"
  672. return " [100%]"
  673. def _write_progress_information_if_past_edge(self) -> None:
  674. w = self._width_of_current_line
  675. if self._show_progress_info == "count":
  676. assert self._session
  677. num_tests = self._session.testscollected
  678. progress_length = len(f" [{num_tests}/{num_tests}]")
  679. elif self._show_progress_info == "times":
  680. progress_length = len(" 99h 59m")
  681. else:
  682. progress_length = len(" [100%]")
  683. past_edge = w + progress_length + 1 >= self._screen_width
  684. if past_edge:
  685. main_color, _ = self._get_main_color()
  686. msg = self._get_progress_information_message()
  687. self._tw.write(msg + "\n", **{main_color: True})
  688. def _write_progress_information_filling_space(self) -> None:
  689. color, _ = self._get_main_color()
  690. msg = self._get_progress_information_message()
  691. w = self._width_of_current_line
  692. fill = self._tw.fullwidth - w - 1
  693. self.write(msg.rjust(fill), flush=True, **{color: True})
  694. @property
  695. def _width_of_current_line(self) -> int:
  696. """Return the width of the current line."""
  697. return self._tw.width_of_current_line
  698. def pytest_collection(self) -> None:
  699. if self.isatty():
  700. if self.config.option.verbose >= 0:
  701. self.write("collecting ... ", flush=True, bold=True)
  702. elif self.config.option.verbose >= 1:
  703. self.write("collecting ... ", flush=True, bold=True)
  704. def pytest_collectreport(self, report: CollectReport) -> None:
  705. if report.failed:
  706. self._add_stats("error", [report])
  707. elif report.skipped:
  708. self._add_stats("skipped", [report])
  709. items = [x for x in report.result if isinstance(x, Item)]
  710. self._numcollected += len(items)
  711. if self.isatty():
  712. self.report_collect()
  713. def report_collect(self, final: bool = False) -> None:
  714. if self.config.option.verbose < 0:
  715. return
  716. if not final:
  717. # Only write the "collecting" report every `REPORT_COLLECTING_RESOLUTION`.
  718. if (
  719. self._collect_report_last_write.elapsed().seconds
  720. < REPORT_COLLECTING_RESOLUTION
  721. ):
  722. return
  723. self._collect_report_last_write = timing.Instant()
  724. errors = len(self.stats.get("error", []))
  725. skipped = len(self.stats.get("skipped", []))
  726. deselected = len(self.stats.get("deselected", []))
  727. selected = self._numcollected - deselected
  728. line = "collected " if final else "collecting "
  729. line += (
  730. str(self._numcollected) + " item" + ("" if self._numcollected == 1 else "s")
  731. )
  732. if errors:
  733. line += f" / {errors} error{'s' if errors != 1 else ''}"
  734. if deselected:
  735. line += f" / {deselected} deselected"
  736. if skipped:
  737. line += f" / {skipped} skipped"
  738. if self._numcollected > selected:
  739. line += f" / {selected} selected"
  740. if self.isatty():
  741. self.rewrite(line, bold=True, erase=True)
  742. if final:
  743. self.write("\n")
  744. else:
  745. self.write_line(line)
  746. @hookimpl(trylast=True)
  747. def pytest_sessionstart(self, session: Session) -> None:
  748. self._session = session
  749. self._session_start = timing.Instant()
  750. if not self.showheader:
  751. return
  752. self.write_sep("=", "test session starts", bold=True)
  753. verinfo = platform.python_version()
  754. if not self.no_header:
  755. msg = f"platform {sys.platform} -- Python {verinfo}"
  756. pypy_version_info = getattr(sys, "pypy_version_info", None)
  757. if pypy_version_info:
  758. verinfo = ".".join(map(str, pypy_version_info[:3]))
  759. msg += f"[pypy-{verinfo}-{pypy_version_info[3]}]"
  760. msg += f", pytest-{_pytest._version.version}, pluggy-{pluggy.__version__}"
  761. if (
  762. self.verbosity > 0
  763. or self.config.option.debug
  764. or getattr(self.config.option, "pastebin", None)
  765. ):
  766. msg += " -- " + str(sys.executable)
  767. self.write_line(msg)
  768. lines = self.config.hook.pytest_report_header(
  769. config=self.config, start_path=self.startpath
  770. )
  771. self._write_report_lines_from_hooks(lines)
  772. def _write_report_lines_from_hooks(
  773. self, lines: Sequence[str | Sequence[str]]
  774. ) -> None:
  775. for line_or_lines in reversed(lines):
  776. if isinstance(line_or_lines, str):
  777. self.write_line(line_or_lines)
  778. else:
  779. for line in line_or_lines:
  780. self.write_line(line)
  781. def pytest_report_header(self, config: Config) -> list[str]:
  782. result = [f"rootdir: {config.rootpath}"]
  783. if config.inipath:
  784. warning = ""
  785. if config._ignored_config_files:
  786. warning = f" (WARNING: ignoring pytest config in {', '.join(config._ignored_config_files)}!)"
  787. result.append(
  788. "configfile: " + bestrelpath(config.rootpath, config.inipath) + warning
  789. )
  790. if config.args_source == Config.ArgsSource.TESTPATHS:
  791. testpaths: list[str] = config.getini("testpaths")
  792. result.append("testpaths: {}".format(", ".join(testpaths)))
  793. plugininfo = config.pluginmanager.list_plugin_distinfo()
  794. if plugininfo:
  795. result.append(
  796. "plugins: {}".format(", ".join(_plugin_nameversions(plugininfo)))
  797. )
  798. return result
  799. def pytest_collection_finish(self, session: Session) -> None:
  800. self.report_collect(True)
  801. lines = self.config.hook.pytest_report_collectionfinish(
  802. config=self.config,
  803. start_path=self.startpath,
  804. items=session.items,
  805. )
  806. self._write_report_lines_from_hooks(lines)
  807. if self.config.getoption("collectonly"):
  808. if session.items:
  809. if self.config.option.verbose > -1:
  810. self._tw.line("")
  811. self._printcollecteditems(session.items)
  812. failed = self.stats.get("failed")
  813. if failed:
  814. self._tw.sep("!", "collection failures")
  815. for rep in failed:
  816. rep.toterminal(self._tw)
  817. def _printcollecteditems(self, items: Sequence[Item]) -> None:
  818. test_cases_verbosity = self.config.get_verbosity(Config.VERBOSITY_TEST_CASES)
  819. if test_cases_verbosity < 0:
  820. if test_cases_verbosity < -1:
  821. counts = Counter(item.nodeid.split("::", 1)[0] for item in items)
  822. for name, count in sorted(counts.items()):
  823. self._tw.line(f"{name}: {count}")
  824. else:
  825. for item in items:
  826. self._tw.line(item.nodeid)
  827. return
  828. stack: list[Node] = []
  829. indent = ""
  830. for item in items:
  831. needed_collectors = item.listchain()[1:] # strip root node
  832. while stack:
  833. if stack == needed_collectors[: len(stack)]:
  834. break
  835. stack.pop()
  836. for col in needed_collectors[len(stack) :]:
  837. stack.append(col)
  838. indent = (len(stack) - 1) * " "
  839. self._tw.line(f"{indent}{col}")
  840. if test_cases_verbosity >= 1:
  841. obj = getattr(col, "obj", None)
  842. doc = inspect.getdoc(obj) if obj else None
  843. if doc:
  844. for line in doc.splitlines():
  845. self._tw.line("{}{}".format(indent + " ", line))
  846. @hookimpl(wrapper=True)
  847. def pytest_sessionfinish(
  848. self, session: Session, exitstatus: int | ExitCode
  849. ) -> Generator[None]:
  850. result = yield
  851. self._tw.line("")
  852. summary_exit_codes = (
  853. ExitCode.OK,
  854. ExitCode.TESTS_FAILED,
  855. ExitCode.INTERRUPTED,
  856. ExitCode.USAGE_ERROR,
  857. ExitCode.NO_TESTS_COLLECTED,
  858. )
  859. if exitstatus in summary_exit_codes and not self.no_summary:
  860. self.config.hook.pytest_terminal_summary(
  861. terminalreporter=self, exitstatus=exitstatus, config=self.config
  862. )
  863. if session.shouldfail:
  864. self.write_sep("!", str(session.shouldfail), red=True)
  865. if exitstatus == ExitCode.INTERRUPTED:
  866. self._report_keyboardinterrupt()
  867. self._keyboardinterrupt_memo = None
  868. elif session.shouldstop:
  869. self.write_sep("!", str(session.shouldstop), red=True)
  870. self.summary_stats()
  871. return result
  872. @hookimpl(wrapper=True)
  873. def pytest_terminal_summary(self) -> Generator[None]:
  874. self.summary_errors()
  875. self.summary_failures()
  876. self.summary_xfailures()
  877. self.summary_warnings()
  878. self.summary_passes()
  879. self.summary_xpasses()
  880. try:
  881. return (yield)
  882. finally:
  883. self.short_test_summary()
  884. # Display any extra warnings from teardown here (if any).
  885. self.summary_warnings()
  886. def pytest_keyboard_interrupt(self, excinfo: ExceptionInfo[BaseException]) -> None:
  887. self._keyboardinterrupt_memo = excinfo.getrepr(funcargs=True)
  888. def pytest_unconfigure(self) -> None:
  889. if self._keyboardinterrupt_memo is not None:
  890. self._report_keyboardinterrupt()
  891. def _report_keyboardinterrupt(self) -> None:
  892. excrepr = self._keyboardinterrupt_memo
  893. assert excrepr is not None
  894. assert excrepr.reprcrash is not None
  895. msg = excrepr.reprcrash.message
  896. self.write_sep("!", msg)
  897. if "KeyboardInterrupt" in msg:
  898. if self.config.option.fulltrace:
  899. excrepr.toterminal(self._tw)
  900. else:
  901. excrepr.reprcrash.toterminal(self._tw)
  902. self._tw.line(
  903. "(to show a full traceback on KeyboardInterrupt use --full-trace)",
  904. yellow=True,
  905. )
  906. def _locationline(
  907. self, nodeid: str, fspath: str, lineno: int | None, domain: str
  908. ) -> str:
  909. def mkrel(nodeid: str) -> str:
  910. line = self.config.cwd_relative_nodeid(nodeid)
  911. if domain and line.endswith(domain):
  912. line = line[: -len(domain)]
  913. values = domain.split("[")
  914. values[0] = values[0].replace(".", "::") # don't replace '.' in params
  915. line += "[".join(values)
  916. return line
  917. # fspath comes from testid which has a "/"-normalized path.
  918. if fspath:
  919. res = mkrel(nodeid)
  920. if self.verbosity >= 2 and nodeid.split("::")[0] != fspath.replace(
  921. "\\", nodes.SEP
  922. ):
  923. res += " <- " + bestrelpath(self.startpath, Path(fspath))
  924. else:
  925. res = "[location]"
  926. return res + " "
  927. def _getfailureheadline(self, rep):
  928. head_line = rep.head_line
  929. if head_line:
  930. return head_line
  931. return "test session" # XXX?
  932. def _getcrashline(self, rep):
  933. try:
  934. return str(rep.longrepr.reprcrash)
  935. except AttributeError:
  936. try:
  937. return str(rep.longrepr)[:50]
  938. except AttributeError:
  939. return ""
  940. #
  941. # Summaries for sessionfinish.
  942. #
  943. def getreports(self, name: str):
  944. return [x for x in self.stats.get(name, ()) if not hasattr(x, "_pdbshown")]
  945. def summary_warnings(self) -> None:
  946. if self.hasopt("w"):
  947. all_warnings: list[WarningReport] | None = self.stats.get("warnings")
  948. if not all_warnings:
  949. return
  950. final = self._already_displayed_warnings is not None
  951. if final:
  952. warning_reports = all_warnings[self._already_displayed_warnings :]
  953. else:
  954. warning_reports = all_warnings
  955. self._already_displayed_warnings = len(warning_reports)
  956. if not warning_reports:
  957. return
  958. reports_grouped_by_message: dict[str, list[WarningReport]] = {}
  959. for wr in warning_reports:
  960. reports_grouped_by_message.setdefault(wr.message, []).append(wr)
  961. def collapsed_location_report(reports: list[WarningReport]) -> str:
  962. locations = []
  963. for w in reports:
  964. location = w.get_location(self.config)
  965. if location:
  966. locations.append(location)
  967. if len(locations) < 10:
  968. return "\n".join(map(str, locations))
  969. counts_by_filename = Counter(
  970. str(loc).split("::", 1)[0] for loc in locations
  971. )
  972. return "\n".join(
  973. "{}: {} warning{}".format(k, v, "s" if v > 1 else "")
  974. for k, v in counts_by_filename.items()
  975. )
  976. title = "warnings summary (final)" if final else "warnings summary"
  977. self.write_sep("=", title, yellow=True, bold=False)
  978. for message, message_reports in reports_grouped_by_message.items():
  979. maybe_location = collapsed_location_report(message_reports)
  980. if maybe_location:
  981. self._tw.line(maybe_location)
  982. lines = message.splitlines()
  983. indented = "\n".join(" " + x for x in lines)
  984. message = indented.rstrip()
  985. else:
  986. message = message.rstrip()
  987. self._tw.line(message)
  988. self._tw.line()
  989. self._tw.line(
  990. "-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html"
  991. )
  992. def summary_passes(self) -> None:
  993. self.summary_passes_combined("passed", "PASSES", "P")
  994. def summary_xpasses(self) -> None:
  995. self.summary_passes_combined("xpassed", "XPASSES", "X")
  996. def summary_passes_combined(
  997. self, which_reports: str, sep_title: str, needed_opt: str
  998. ) -> None:
  999. if self.config.option.tbstyle != "no":
  1000. if self.hasopt(needed_opt):
  1001. reports: list[TestReport] = self.getreports(which_reports)
  1002. if not reports:
  1003. return
  1004. self.write_sep("=", sep_title)
  1005. for rep in reports:
  1006. if rep.sections:
  1007. msg = self._getfailureheadline(rep)
  1008. self.write_sep("_", msg, green=True, bold=True)
  1009. self._outrep_summary(rep)
  1010. self._handle_teardown_sections(rep.nodeid)
  1011. def _get_teardown_reports(self, nodeid: str) -> list[TestReport]:
  1012. reports = self.getreports("")
  1013. return [
  1014. report
  1015. for report in reports
  1016. if report.when == "teardown" and report.nodeid == nodeid
  1017. ]
  1018. def _handle_teardown_sections(self, nodeid: str) -> None:
  1019. for report in self._get_teardown_reports(nodeid):
  1020. self.print_teardown_sections(report)
  1021. def print_teardown_sections(self, rep: TestReport) -> None:
  1022. showcapture = self.config.option.showcapture
  1023. if showcapture == "no":
  1024. return
  1025. for secname, content in rep.sections:
  1026. if showcapture != "all" and showcapture not in secname:
  1027. continue
  1028. if "teardown" in secname:
  1029. self._tw.sep("-", secname)
  1030. if content[-1:] == "\n":
  1031. content = content[:-1]
  1032. self._tw.line(content)
  1033. def summary_failures(self) -> None:
  1034. style = self.config.option.tbstyle
  1035. self.summary_failures_combined("failed", "FAILURES", style=style)
  1036. def summary_xfailures(self) -> None:
  1037. show_tb = self.config.option.xfail_tb
  1038. style = self.config.option.tbstyle if show_tb else "no"
  1039. self.summary_failures_combined("xfailed", "XFAILURES", style=style)
  1040. def summary_failures_combined(
  1041. self,
  1042. which_reports: str,
  1043. sep_title: str,
  1044. *,
  1045. style: str,
  1046. needed_opt: str | None = None,
  1047. ) -> None:
  1048. if style != "no":
  1049. if not needed_opt or self.hasopt(needed_opt):
  1050. reports: list[BaseReport] = self.getreports(which_reports)
  1051. if not reports:
  1052. return
  1053. self.write_sep("=", sep_title)
  1054. if style == "line":
  1055. for rep in reports:
  1056. line = self._getcrashline(rep)
  1057. self._outrep_summary(rep)
  1058. self.write_line(line)
  1059. else:
  1060. for rep in reports:
  1061. msg = self._getfailureheadline(rep)
  1062. self.write_sep("_", msg, red=True, bold=True)
  1063. self._outrep_summary(rep)
  1064. self._handle_teardown_sections(rep.nodeid)
  1065. def summary_errors(self) -> None:
  1066. if self.config.option.tbstyle != "no":
  1067. reports: list[BaseReport] = self.getreports("error")
  1068. if not reports:
  1069. return
  1070. self.write_sep("=", "ERRORS")
  1071. for rep in self.stats["error"]:
  1072. msg = self._getfailureheadline(rep)
  1073. if rep.when == "collect":
  1074. msg = "ERROR collecting " + msg
  1075. else:
  1076. msg = f"ERROR at {rep.when} of {msg}"
  1077. self.write_sep("_", msg, red=True, bold=True)
  1078. self._outrep_summary(rep)
  1079. def _outrep_summary(self, rep: BaseReport) -> None:
  1080. rep.toterminal(self._tw)
  1081. showcapture = self.config.option.showcapture
  1082. if showcapture == "no":
  1083. return
  1084. for secname, content in rep.sections:
  1085. if showcapture != "all" and showcapture not in secname:
  1086. continue
  1087. self._tw.sep("-", secname)
  1088. if content[-1:] == "\n":
  1089. content = content[:-1]
  1090. self._tw.line(content)
  1091. def summary_stats(self) -> None:
  1092. if self.verbosity < -1:
  1093. return
  1094. session_duration = self._session_start.elapsed()
  1095. (parts, main_color) = self.build_summary_stats_line()
  1096. line_parts = []
  1097. display_sep = self.verbosity >= 0
  1098. if display_sep:
  1099. fullwidth = self._tw.fullwidth
  1100. for text, markup in parts:
  1101. with_markup = self._tw.markup(text, **markup)
  1102. if display_sep:
  1103. fullwidth += len(with_markup) - len(text)
  1104. line_parts.append(with_markup)
  1105. msg = ", ".join(line_parts)
  1106. main_markup = {main_color: True}
  1107. duration = f" in {format_session_duration(session_duration.seconds)}"
  1108. duration_with_markup = self._tw.markup(duration, **main_markup)
  1109. if display_sep:
  1110. fullwidth += len(duration_with_markup) - len(duration)
  1111. msg += duration_with_markup
  1112. if display_sep:
  1113. markup_for_end_sep = self._tw.markup("", **main_markup)
  1114. if markup_for_end_sep.endswith("\x1b[0m"):
  1115. markup_for_end_sep = markup_for_end_sep[:-4]
  1116. fullwidth += len(markup_for_end_sep)
  1117. msg += markup_for_end_sep
  1118. if display_sep:
  1119. self.write_sep("=", msg, fullwidth=fullwidth, **main_markup)
  1120. else:
  1121. self.write_line(msg, **main_markup)
  1122. def short_test_summary(self) -> None:
  1123. if not self.reportchars:
  1124. return
  1125. def show_simple(lines: list[str], *, stat: str) -> None:
  1126. failed = self.stats.get(stat, [])
  1127. if not failed:
  1128. return
  1129. config = self.config
  1130. for rep in failed:
  1131. color = _color_for_type.get(stat, _color_for_type_default)
  1132. line = _get_line_with_reprcrash_message(
  1133. config, rep, self._tw, {color: True}
  1134. )
  1135. lines.append(line)
  1136. def show_xfailed(lines: list[str]) -> None:
  1137. xfailed = self.stats.get("xfailed", [])
  1138. for rep in xfailed:
  1139. verbose_word, verbose_markup = rep._get_verbose_word_with_markup(
  1140. self.config, {_color_for_type["warnings"]: True}
  1141. )
  1142. markup_word = self._tw.markup(verbose_word, **verbose_markup)
  1143. nodeid = _get_node_id_with_markup(self._tw, self.config, rep)
  1144. line = f"{markup_word} {nodeid}"
  1145. reason = rep.wasxfail
  1146. if reason:
  1147. line += " - " + str(reason)
  1148. lines.append(line)
  1149. def show_xpassed(lines: list[str]) -> None:
  1150. xpassed = self.stats.get("xpassed", [])
  1151. for rep in xpassed:
  1152. verbose_word, verbose_markup = rep._get_verbose_word_with_markup(
  1153. self.config, {_color_for_type["warnings"]: True}
  1154. )
  1155. markup_word = self._tw.markup(verbose_word, **verbose_markup)
  1156. nodeid = _get_node_id_with_markup(self._tw, self.config, rep)
  1157. line = f"{markup_word} {nodeid}"
  1158. reason = rep.wasxfail
  1159. if reason:
  1160. line += " - " + str(reason)
  1161. lines.append(line)
  1162. def show_skipped_folded(lines: list[str]) -> None:
  1163. skipped: list[CollectReport] = self.stats.get("skipped", [])
  1164. fskips = _folded_skips(self.startpath, skipped) if skipped else []
  1165. if not fskips:
  1166. return
  1167. verbose_word, verbose_markup = skipped[0]._get_verbose_word_with_markup(
  1168. self.config, {_color_for_type["warnings"]: True}
  1169. )
  1170. markup_word = self._tw.markup(verbose_word, **verbose_markup)
  1171. prefix = "Skipped: "
  1172. for num, fspath, lineno, reason in fskips:
  1173. if reason.startswith(prefix):
  1174. reason = reason[len(prefix) :]
  1175. if lineno is not None:
  1176. lines.append(f"{markup_word} [{num}] {fspath}:{lineno}: {reason}")
  1177. else:
  1178. lines.append(f"{markup_word} [{num}] {fspath}: {reason}")
  1179. def show_skipped_unfolded(lines: list[str]) -> None:
  1180. skipped: list[CollectReport] = self.stats.get("skipped", [])
  1181. for rep in skipped:
  1182. assert rep.longrepr is not None
  1183. assert isinstance(rep.longrepr, tuple), (rep, rep.longrepr)
  1184. assert len(rep.longrepr) == 3, (rep, rep.longrepr)
  1185. verbose_word, verbose_markup = rep._get_verbose_word_with_markup(
  1186. self.config, {_color_for_type["warnings"]: True}
  1187. )
  1188. markup_word = self._tw.markup(verbose_word, **verbose_markup)
  1189. nodeid = _get_node_id_with_markup(self._tw, self.config, rep)
  1190. line = f"{markup_word} {nodeid}"
  1191. reason = rep.longrepr[2]
  1192. if reason:
  1193. line += " - " + str(reason)
  1194. lines.append(line)
  1195. def show_skipped(lines: list[str]) -> None:
  1196. if self.foldskipped:
  1197. show_skipped_folded(lines)
  1198. else:
  1199. show_skipped_unfolded(lines)
  1200. REPORTCHAR_ACTIONS: Mapping[str, Callable[[list[str]], None]] = {
  1201. "x": show_xfailed,
  1202. "X": show_xpassed,
  1203. "f": partial(show_simple, stat="failed"),
  1204. "s": show_skipped,
  1205. "p": partial(show_simple, stat="passed"),
  1206. "E": partial(show_simple, stat="error"),
  1207. }
  1208. lines: list[str] = []
  1209. for char in self.reportchars:
  1210. action = REPORTCHAR_ACTIONS.get(char)
  1211. if action: # skipping e.g. "P" (passed with output) here.
  1212. action(lines)
  1213. if lines:
  1214. self.write_sep("=", "short test summary info", cyan=True, bold=True)
  1215. for line in lines:
  1216. self.write_line(line)
  1217. def _get_main_color(self) -> tuple[str, list[str]]:
  1218. if self._main_color is None or self._known_types is None or self._is_last_item:
  1219. self._set_main_color()
  1220. assert self._main_color
  1221. assert self._known_types
  1222. return self._main_color, self._known_types
  1223. def _determine_main_color(self, unknown_type_seen: bool) -> str:
  1224. stats = self.stats
  1225. if "failed" in stats or "error" in stats:
  1226. main_color = "red"
  1227. elif "warnings" in stats or "xpassed" in stats or unknown_type_seen:
  1228. main_color = "yellow"
  1229. elif "passed" in stats or not self._is_last_item:
  1230. main_color = "green"
  1231. else:
  1232. main_color = "yellow"
  1233. return main_color
  1234. def _set_main_color(self) -> None:
  1235. unknown_types: list[str] = []
  1236. for found_type in self.stats:
  1237. if found_type: # setup/teardown reports have an empty key, ignore them
  1238. if found_type not in KNOWN_TYPES and found_type not in unknown_types:
  1239. unknown_types.append(found_type)
  1240. self._known_types = list(KNOWN_TYPES) + unknown_types
  1241. self._main_color = self._determine_main_color(bool(unknown_types))
  1242. def build_summary_stats_line(self) -> tuple[list[tuple[str, dict[str, bool]]], str]:
  1243. """
  1244. Build the parts used in the last summary stats line.
  1245. The summary stats line is the line shown at the end, "=== 12 passed, 2 errors in Xs===".
  1246. This function builds a list of the "parts" that make up for the text in that line, in
  1247. the example above it would be::
  1248. [
  1249. ("12 passed", {"green": True}),
  1250. ("2 errors", {"red": True}
  1251. ]
  1252. That last dict for each line is a "markup dictionary", used by TerminalWriter to
  1253. color output.
  1254. The final color of the line is also determined by this function, and is the second
  1255. element of the returned tuple.
  1256. """
  1257. if self.config.getoption("collectonly"):
  1258. return self._build_collect_only_summary_stats_line()
  1259. else:
  1260. return self._build_normal_summary_stats_line()
  1261. def _get_reports_to_display(self, key: str) -> list[Any]:
  1262. """Get test/collection reports for the given status key, such as `passed` or `error`."""
  1263. reports = self.stats.get(key, [])
  1264. return [x for x in reports if getattr(x, "count_towards_summary", True)]
  1265. def _build_normal_summary_stats_line(
  1266. self,
  1267. ) -> tuple[list[tuple[str, dict[str, bool]]], str]:
  1268. main_color, known_types = self._get_main_color()
  1269. parts = []
  1270. for key in known_types:
  1271. reports = self._get_reports_to_display(key)
  1272. if reports:
  1273. count = len(reports)
  1274. color = _color_for_type.get(key, _color_for_type_default)
  1275. markup = {color: True, "bold": color == main_color}
  1276. parts.append(("%d %s" % pluralize(count, key), markup)) # noqa: UP031
  1277. if not parts:
  1278. parts = [("no tests ran", {_color_for_type_default: True})]
  1279. return parts, main_color
  1280. def _build_collect_only_summary_stats_line(
  1281. self,
  1282. ) -> tuple[list[tuple[str, dict[str, bool]]], str]:
  1283. deselected = len(self._get_reports_to_display("deselected"))
  1284. errors = len(self._get_reports_to_display("error"))
  1285. if self._numcollected == 0:
  1286. parts = [("no tests collected", {"yellow": True})]
  1287. main_color = "yellow"
  1288. elif deselected == 0:
  1289. main_color = "green"
  1290. collected_output = "%d %s collected" % pluralize(self._numcollected, "test") # noqa: UP031
  1291. parts = [(collected_output, {main_color: True})]
  1292. else:
  1293. all_tests_were_deselected = self._numcollected == deselected
  1294. if all_tests_were_deselected:
  1295. main_color = "yellow"
  1296. collected_output = f"no tests collected ({deselected} deselected)"
  1297. else:
  1298. main_color = "green"
  1299. selected = self._numcollected - deselected
  1300. collected_output = f"{selected}/{self._numcollected} tests collected ({deselected} deselected)"
  1301. parts = [(collected_output, {main_color: True})]
  1302. if errors:
  1303. main_color = _color_for_type["error"]
  1304. parts += [("%d %s" % pluralize(errors, "error"), {main_color: True})] # noqa: UP031
  1305. return parts, main_color
  1306. def _get_node_id_with_markup(tw: TerminalWriter, config: Config, rep: BaseReport):
  1307. nodeid = config.cwd_relative_nodeid(rep.nodeid)
  1308. path, *parts = nodeid.split("::")
  1309. if parts:
  1310. parts_markup = tw.markup("::".join(parts), bold=True)
  1311. return path + "::" + parts_markup
  1312. else:
  1313. return path
  1314. def _format_trimmed(format: str, msg: str, available_width: int) -> str | None:
  1315. """Format msg into format, ellipsizing it if doesn't fit in available_width.
  1316. Returns None if even the ellipsis can't fit.
  1317. """
  1318. # Only use the first line.
  1319. i = msg.find("\n")
  1320. if i != -1:
  1321. msg = msg[:i]
  1322. ellipsis = "..."
  1323. format_width = wcswidth(format.format(""))
  1324. if format_width + len(ellipsis) > available_width:
  1325. return None
  1326. if format_width + wcswidth(msg) > available_width:
  1327. available_width -= len(ellipsis)
  1328. msg = msg[:available_width]
  1329. while format_width + wcswidth(msg) > available_width:
  1330. msg = msg[:-1]
  1331. msg += ellipsis
  1332. return format.format(msg)
  1333. def _get_line_with_reprcrash_message(
  1334. config: Config, rep: BaseReport, tw: TerminalWriter, word_markup: dict[str, bool]
  1335. ) -> str:
  1336. """Get summary line for a report, trying to add reprcrash message."""
  1337. verbose_word, verbose_markup = rep._get_verbose_word_with_markup(
  1338. config, word_markup
  1339. )
  1340. word = tw.markup(verbose_word, **verbose_markup)
  1341. node = _get_node_id_with_markup(tw, config, rep)
  1342. line = f"{word} {node}"
  1343. line_width = wcswidth(line)
  1344. msg: str | None
  1345. try:
  1346. if isinstance(rep.longrepr, str):
  1347. msg = rep.longrepr
  1348. else:
  1349. # Type ignored intentionally -- possible AttributeError expected.
  1350. msg = rep.longrepr.reprcrash.message # type: ignore[union-attr]
  1351. except AttributeError:
  1352. pass
  1353. else:
  1354. if (
  1355. running_on_ci() or config.option.verbose >= 2
  1356. ) and not config.option.force_short_summary:
  1357. msg = f" - {msg}"
  1358. else:
  1359. available_width = tw.fullwidth - line_width
  1360. msg = _format_trimmed(" - {}", msg, available_width)
  1361. if msg is not None:
  1362. line += msg
  1363. return line
  1364. def _folded_skips(
  1365. startpath: Path,
  1366. skipped: Sequence[CollectReport],
  1367. ) -> list[tuple[int, str, int | None, str]]:
  1368. d: dict[tuple[str, int | None, str], list[CollectReport]] = {}
  1369. for event in skipped:
  1370. assert event.longrepr is not None
  1371. assert isinstance(event.longrepr, tuple), (event, event.longrepr)
  1372. assert len(event.longrepr) == 3, (event, event.longrepr)
  1373. fspath, lineno, reason = event.longrepr
  1374. # For consistency, report all fspaths in relative form.
  1375. fspath = bestrelpath(startpath, Path(fspath))
  1376. keywords = getattr(event, "keywords", {})
  1377. # Folding reports with global pytestmark variable.
  1378. # This is a workaround, because for now we cannot identify the scope of a skip marker
  1379. # TODO: Revisit after marks scope would be fixed.
  1380. if (
  1381. event.when == "setup"
  1382. and "skip" in keywords
  1383. and "pytestmark" not in keywords
  1384. ):
  1385. key: tuple[str, int | None, str] = (fspath, None, reason)
  1386. else:
  1387. key = (fspath, lineno, reason)
  1388. d.setdefault(key, []).append(event)
  1389. values: list[tuple[int, str, int | None, str]] = []
  1390. for key, events in d.items():
  1391. values.append((len(events), *key))
  1392. return values
  1393. _color_for_type = {
  1394. "failed": "red",
  1395. "error": "red",
  1396. "warnings": "yellow",
  1397. "passed": "green",
  1398. "subtests passed": "green",
  1399. "subtests failed": "red",
  1400. }
  1401. _color_for_type_default = "yellow"
  1402. def pluralize(count: int, noun: str) -> tuple[int, str]:
  1403. # No need to pluralize words such as `failed` or `passed`.
  1404. if noun not in ["error", "warnings", "test"]:
  1405. return count, noun
  1406. # The `warnings` key is plural. To avoid API breakage, we keep it that way but
  1407. # set it to singular here so we can determine plurality in the same way as we do
  1408. # for `error`.
  1409. noun = noun.replace("warnings", "warning")
  1410. return count, noun + "s" if count != 1 else noun
  1411. def _plugin_nameversions(plugininfo) -> list[str]:
  1412. values: list[str] = []
  1413. for plugin, dist in plugininfo:
  1414. # Gets us name and version!
  1415. name = f"{dist.project_name}-{dist.version}"
  1416. # Questionable convenience, but it keeps things short.
  1417. if name.startswith("pytest-"):
  1418. name = name[7:]
  1419. # We decided to print python package names they can have more than one plugin.
  1420. if name not in values:
  1421. values.append(name)
  1422. return values
  1423. def format_session_duration(seconds: float) -> str:
  1424. """Format the given seconds in a human readable manner to show in the final summary."""
  1425. if seconds < 60:
  1426. return f"{seconds:.2f}s"
  1427. else:
  1428. dt = datetime.timedelta(seconds=int(seconds))
  1429. return f"{seconds:.2f}s ({dt})"
  1430. def format_node_duration(seconds: float) -> str:
  1431. """Format the given seconds in a human readable manner to show in the test progress."""
  1432. # The formatting is designed to be compact and readable, with at most 7 characters
  1433. # for durations below 100 hours.
  1434. if seconds < 0.00001:
  1435. return f" {seconds * 1000000:.3f}us"
  1436. if seconds < 0.0001:
  1437. return f" {seconds * 1000000:.2f}us"
  1438. if seconds < 0.001:
  1439. return f" {seconds * 1000000:.1f}us"
  1440. if seconds < 0.01:
  1441. return f" {seconds * 1000:.3f}ms"
  1442. if seconds < 0.1:
  1443. return f" {seconds * 1000:.2f}ms"
  1444. if seconds < 1:
  1445. return f" {seconds * 1000:.1f}ms"
  1446. if seconds < 60:
  1447. return f" {seconds:.3f}s"
  1448. if seconds < 3600:
  1449. return f" {seconds // 60:.0f}m {seconds % 60:.0f}s"
  1450. return f" {seconds // 3600:.0f}h {(seconds % 3600) // 60:.0f}m"
  1451. def _get_raw_skip_reason(report: TestReport) -> str:
  1452. """Get the reason string of a skip/xfail/xpass test report.
  1453. The string is just the part given by the user.
  1454. """
  1455. if hasattr(report, "wasxfail"):
  1456. reason = report.wasxfail
  1457. if reason.startswith("reason: "):
  1458. reason = reason[len("reason: ") :]
  1459. return reason
  1460. else:
  1461. assert report.skipped
  1462. assert isinstance(report.longrepr, tuple)
  1463. _, _, reason = report.longrepr
  1464. if reason.startswith("Skipped: "):
  1465. reason = reason[len("Skipped: ") :]
  1466. elif reason == "Skipped":
  1467. reason = ""
  1468. return reason
  1469. class TerminalProgressPlugin:
  1470. """Terminal progress reporting plugin using OSC 9;4 ANSI sequences.
  1471. Emits OSC 9;4 sequences to indicate test progress to terminal
  1472. tabs/windows/etc.
  1473. Not all terminal emulators support this feature.
  1474. Ref: https://conemu.github.io/en/AnsiEscapeCodes.html#ConEmu_specific_OSC
  1475. """
  1476. def __init__(self, tr: TerminalReporter) -> None:
  1477. self._tr = tr
  1478. self._session: Session | None = None
  1479. self._has_failures = False
  1480. def _emit_progress(
  1481. self,
  1482. state: Literal["remove", "normal", "error", "indeterminate", "paused"],
  1483. progress: int | None = None,
  1484. ) -> None:
  1485. """Emit OSC 9;4 sequence for indicating progress to the terminal.
  1486. :param state:
  1487. Progress state to set.
  1488. :param progress:
  1489. Progress value 0-100. Required for "normal", optional for "error"
  1490. and "paused", otherwise ignored.
  1491. """
  1492. assert progress is None or 0 <= progress <= 100
  1493. # OSC 9;4 sequence: ESC ] 9 ; 4 ; state ; progress ST
  1494. # ST can be ESC \ or BEL. ESC \ seems better supported.
  1495. match state:
  1496. case "remove":
  1497. sequence = "\x1b]9;4;0;\x1b\\"
  1498. case "normal":
  1499. assert progress is not None
  1500. sequence = f"\x1b]9;4;1;{progress}\x1b\\"
  1501. case "error":
  1502. if progress is not None:
  1503. sequence = f"\x1b]9;4;2;{progress}\x1b\\"
  1504. else:
  1505. sequence = "\x1b]9;4;2;\x1b\\"
  1506. case "indeterminate":
  1507. sequence = "\x1b]9;4;3;\x1b\\"
  1508. case "paused":
  1509. if progress is not None:
  1510. sequence = f"\x1b]9;4;4;{progress}\x1b\\"
  1511. else:
  1512. sequence = "\x1b]9;4;4;\x1b\\"
  1513. self._tr.write_raw(sequence, flush=True)
  1514. @hookimpl
  1515. def pytest_sessionstart(self, session: Session) -> None:
  1516. self._session = session
  1517. # Show indeterminate progress during collection.
  1518. self._emit_progress("indeterminate")
  1519. @hookimpl
  1520. def pytest_collection_finish(self) -> None:
  1521. assert self._session is not None
  1522. if self._session.testscollected > 0:
  1523. # Switch from indeterminate to 0% progress.
  1524. self._emit_progress("normal", 0)
  1525. @hookimpl
  1526. def pytest_runtest_logreport(self, report: TestReport) -> None:
  1527. if report.failed:
  1528. self._has_failures = True
  1529. # Let's consider the "call" phase for progress.
  1530. if report.when != "call":
  1531. return
  1532. # Calculate and emit progress.
  1533. assert self._session is not None
  1534. collected = self._session.testscollected
  1535. if collected > 0:
  1536. reported = self._tr.reported_progress
  1537. progress = min(reported * 100 // collected, 100)
  1538. self._emit_progress("error" if self._has_failures else "normal", progress)
  1539. @hookimpl
  1540. def pytest_sessionfinish(self) -> None:
  1541. self._emit_progress("remove")