subtests.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411
  1. """Builtin plugin that adds subtests support."""
  2. from __future__ import annotations
  3. from collections import defaultdict
  4. from collections.abc import Callable
  5. from collections.abc import Iterator
  6. from collections.abc import Mapping
  7. from contextlib import AbstractContextManager
  8. from contextlib import contextmanager
  9. from contextlib import ExitStack
  10. from contextlib import nullcontext
  11. import dataclasses
  12. import time
  13. from types import TracebackType
  14. from typing import Any
  15. from typing import TYPE_CHECKING
  16. import pluggy
  17. from _pytest._code import ExceptionInfo
  18. from _pytest._io.saferepr import saferepr
  19. from _pytest.capture import CaptureFixture
  20. from _pytest.capture import FDCapture
  21. from _pytest.capture import SysCapture
  22. from _pytest.config import Config
  23. from _pytest.config import hookimpl
  24. from _pytest.config.argparsing import Parser
  25. from _pytest.deprecated import check_ispytest
  26. from _pytest.fixtures import fixture
  27. from _pytest.fixtures import SubRequest
  28. from _pytest.logging import catching_logs
  29. from _pytest.logging import LogCaptureHandler
  30. from _pytest.logging import LoggingPlugin
  31. from _pytest.reports import TestReport
  32. from _pytest.runner import CallInfo
  33. from _pytest.runner import check_interactive_exception
  34. from _pytest.runner import get_reraise_exceptions
  35. from _pytest.stash import StashKey
  36. if TYPE_CHECKING:
  37. from typing_extensions import Self
  38. def pytest_addoption(parser: Parser) -> None:
  39. Config._add_verbosity_ini(
  40. parser,
  41. Config.VERBOSITY_SUBTESTS,
  42. help=(
  43. "Specify verbosity level for subtests. "
  44. "Higher levels will generate output for passed subtests. Failed subtests are always reported."
  45. ),
  46. )
  47. @dataclasses.dataclass(frozen=True, slots=True, kw_only=True)
  48. class SubtestContext:
  49. """The values passed to Subtests.test() that are included in the test report."""
  50. msg: str | None
  51. kwargs: Mapping[str, Any]
  52. def _to_json(self) -> dict[str, Any]:
  53. return dataclasses.asdict(self)
  54. @classmethod
  55. def _from_json(cls, d: dict[str, Any]) -> Self:
  56. return cls(msg=d["msg"], kwargs=d["kwargs"])
  57. @dataclasses.dataclass(init=False)
  58. class SubtestReport(TestReport):
  59. context: SubtestContext
  60. @property
  61. def head_line(self) -> str:
  62. _, _, domain = self.location
  63. return f"{domain} {self._sub_test_description()}"
  64. def _sub_test_description(self) -> str:
  65. parts = []
  66. if self.context.msg is not None:
  67. parts.append(f"[{self.context.msg}]")
  68. if self.context.kwargs:
  69. params_desc = ", ".join(
  70. f"{k}={saferepr(v)}" for (k, v) in self.context.kwargs.items()
  71. )
  72. parts.append(f"({params_desc})")
  73. return " ".join(parts) or "(<subtest>)"
  74. def _to_json(self) -> dict[str, Any]:
  75. data = super()._to_json()
  76. del data["context"]
  77. data["_report_type"] = "SubTestReport"
  78. data["_subtest.context"] = self.context._to_json()
  79. return data
  80. @classmethod
  81. def _from_json(cls, reportdict: dict[str, Any]) -> SubtestReport:
  82. report = super()._from_json(reportdict)
  83. report.context = SubtestContext._from_json(reportdict["_subtest.context"])
  84. return report
  85. @classmethod
  86. def _new(
  87. cls,
  88. test_report: TestReport,
  89. context: SubtestContext,
  90. captured_output: Captured | None,
  91. captured_logs: CapturedLogs | None,
  92. ) -> Self:
  93. result = super()._from_json(test_report._to_json())
  94. result.context = context
  95. if captured_output:
  96. if captured_output.out:
  97. result.sections.append(("Captured stdout call", captured_output.out))
  98. if captured_output.err:
  99. result.sections.append(("Captured stderr call", captured_output.err))
  100. if captured_logs and (log := captured_logs.handler.stream.getvalue()):
  101. result.sections.append(("Captured log call", log))
  102. return result
  103. @fixture
  104. def subtests(request: SubRequest) -> Subtests:
  105. """Provides subtests functionality."""
  106. capmam = request.node.config.pluginmanager.get_plugin("capturemanager")
  107. suspend_capture_ctx = (
  108. capmam.global_and_fixture_disabled if capmam is not None else nullcontext
  109. )
  110. return Subtests(request.node.ihook, suspend_capture_ctx, request, _ispytest=True)
  111. class Subtests:
  112. """Subtests fixture, enables declaring subtests inside test functions via the :meth:`test` method."""
  113. def __init__(
  114. self,
  115. ihook: pluggy.HookRelay,
  116. suspend_capture_ctx: Callable[[], AbstractContextManager[None]],
  117. request: SubRequest,
  118. *,
  119. _ispytest: bool = False,
  120. ) -> None:
  121. check_ispytest(_ispytest)
  122. self._ihook = ihook
  123. self._suspend_capture_ctx = suspend_capture_ctx
  124. self._request = request
  125. def test(
  126. self,
  127. msg: str | None = None,
  128. **kwargs: Any,
  129. ) -> _SubTestContextManager:
  130. """
  131. Context manager for subtests, capturing exceptions raised inside the subtest scope and
  132. reporting assertion failures and errors individually.
  133. Usage
  134. -----
  135. .. code-block:: python
  136. def test(subtests):
  137. for i in range(5):
  138. with subtests.test("custom message", i=i):
  139. assert i % 2 == 0
  140. :param msg:
  141. If given, the message will be shown in the test report in case of subtest failure.
  142. :param kwargs:
  143. Arbitrary values that are also added to the subtest report.
  144. """
  145. return _SubTestContextManager(
  146. self._ihook,
  147. msg,
  148. kwargs,
  149. request=self._request,
  150. suspend_capture_ctx=self._suspend_capture_ctx,
  151. config=self._request.config,
  152. )
  153. @dataclasses.dataclass
  154. class _SubTestContextManager:
  155. """
  156. Context manager for subtests, capturing exceptions raised inside the subtest scope and handling
  157. them through the pytest machinery.
  158. """
  159. # Note: initially the logic for this context manager was implemented directly
  160. # in Subtests.test() as a @contextmanager, however, it is not possible to control the output fully when
  161. # exiting from it due to an exception when in `--exitfirst` mode, so this was refactored into an
  162. # explicit context manager class (pytest-dev/pytest-subtests#134).
  163. ihook: pluggy.HookRelay
  164. msg: str | None
  165. kwargs: dict[str, Any]
  166. suspend_capture_ctx: Callable[[], AbstractContextManager[None]]
  167. request: SubRequest
  168. config: Config
  169. def __enter__(self) -> None:
  170. __tracebackhide__ = True
  171. self._start = time.time()
  172. self._precise_start = time.perf_counter()
  173. self._exc_info = None
  174. self._exit_stack = ExitStack()
  175. self._captured_output = self._exit_stack.enter_context(
  176. capturing_output(self.request)
  177. )
  178. self._captured_logs = self._exit_stack.enter_context(
  179. capturing_logs(self.request)
  180. )
  181. def __exit__(
  182. self,
  183. exc_type: type[BaseException] | None,
  184. exc_val: BaseException | None,
  185. exc_tb: TracebackType | None,
  186. ) -> bool:
  187. __tracebackhide__ = True
  188. if exc_val is not None:
  189. exc_info = ExceptionInfo.from_exception(exc_val)
  190. else:
  191. exc_info = None
  192. self._exit_stack.close()
  193. precise_stop = time.perf_counter()
  194. duration = precise_stop - self._precise_start
  195. stop = time.time()
  196. call_info = CallInfo[None](
  197. None,
  198. exc_info,
  199. start=self._start,
  200. stop=stop,
  201. duration=duration,
  202. when="call",
  203. _ispytest=True,
  204. )
  205. report = self.ihook.pytest_runtest_makereport(
  206. item=self.request.node, call=call_info
  207. )
  208. sub_report = SubtestReport._new(
  209. report,
  210. SubtestContext(msg=self.msg, kwargs=self.kwargs),
  211. captured_output=self._captured_output,
  212. captured_logs=self._captured_logs,
  213. )
  214. if sub_report.failed:
  215. failed_subtests = self.config.stash[failed_subtests_key]
  216. failed_subtests[self.request.node.nodeid] += 1
  217. with self.suspend_capture_ctx():
  218. self.ihook.pytest_runtest_logreport(report=sub_report)
  219. if check_interactive_exception(call_info, sub_report):
  220. self.ihook.pytest_exception_interact(
  221. node=self.request.node, call=call_info, report=sub_report
  222. )
  223. if exc_val is not None:
  224. if isinstance(exc_val, get_reraise_exceptions(self.config)):
  225. return False
  226. if self.request.session.shouldfail:
  227. return False
  228. return True
  229. @contextmanager
  230. def capturing_output(request: SubRequest) -> Iterator[Captured]:
  231. option = request.config.getoption("capture", None)
  232. capman = request.config.pluginmanager.getplugin("capturemanager")
  233. if getattr(capman, "_capture_fixture", None):
  234. # capsys or capfd are active, subtest should not capture.
  235. fixture = None
  236. elif option == "sys":
  237. fixture = CaptureFixture(SysCapture, request, _ispytest=True)
  238. elif option == "fd":
  239. fixture = CaptureFixture(FDCapture, request, _ispytest=True)
  240. else:
  241. fixture = None
  242. if fixture is not None:
  243. fixture._start()
  244. captured = Captured()
  245. try:
  246. yield captured
  247. finally:
  248. if fixture is not None:
  249. out, err = fixture.readouterr()
  250. fixture.close()
  251. captured.out = out
  252. captured.err = err
  253. @contextmanager
  254. def capturing_logs(
  255. request: SubRequest,
  256. ) -> Iterator[CapturedLogs | None]:
  257. logging_plugin: LoggingPlugin | None = request.config.pluginmanager.getplugin(
  258. "logging-plugin"
  259. )
  260. if logging_plugin is None:
  261. yield None
  262. else:
  263. handler = LogCaptureHandler()
  264. handler.setFormatter(logging_plugin.formatter)
  265. captured_logs = CapturedLogs(handler)
  266. with catching_logs(handler, level=logging_plugin.log_level):
  267. yield captured_logs
  268. @dataclasses.dataclass
  269. class Captured:
  270. out: str = ""
  271. err: str = ""
  272. @dataclasses.dataclass
  273. class CapturedLogs:
  274. handler: LogCaptureHandler
  275. def pytest_report_to_serializable(report: TestReport) -> dict[str, Any] | None:
  276. if isinstance(report, SubtestReport):
  277. return report._to_json()
  278. return None
  279. def pytest_report_from_serializable(data: dict[str, Any]) -> SubtestReport | None:
  280. if data.get("_report_type") == "SubTestReport":
  281. return SubtestReport._from_json(data)
  282. return None
  283. # Dict of nodeid -> number of failed subtests.
  284. # Used to fail top-level tests that passed but contain failed subtests.
  285. failed_subtests_key = StashKey[defaultdict[str, int]]()
  286. def pytest_configure(config: Config) -> None:
  287. config.stash[failed_subtests_key] = defaultdict(lambda: 0)
  288. @hookimpl(tryfirst=True)
  289. def pytest_report_teststatus(
  290. report: TestReport,
  291. config: Config,
  292. ) -> tuple[str, str, str | Mapping[str, bool]] | None:
  293. if report.when != "call":
  294. return None
  295. quiet = config.get_verbosity(Config.VERBOSITY_SUBTESTS) == 0
  296. if isinstance(report, SubtestReport):
  297. outcome = report.outcome
  298. description = report._sub_test_description()
  299. if hasattr(report, "wasxfail"):
  300. if quiet:
  301. return "", "", ""
  302. elif outcome == "skipped":
  303. category = "xfailed"
  304. short = "y" # x letter is used for regular xfail, y for subtest xfail
  305. status = "SUBXFAIL"
  306. # outcome == "passed" in an xfail is only possible via a @pytest.mark.xfail mark, which
  307. # is not applicable to a subtest, which only handles pytest.xfail().
  308. else: # pragma: no cover
  309. # This should not normally happen, unless some plugin is setting wasxfail without
  310. # the correct outcome. Pytest expects the call outcome to be either skipped or
  311. # passed in case of xfail.
  312. # Let's pass this report to the next hook.
  313. return None
  314. return category, short, f"{status}{description}"
  315. if report.failed:
  316. return outcome, "u", f"SUBFAILED{description}"
  317. else:
  318. if report.passed:
  319. if quiet:
  320. return "", "", ""
  321. else:
  322. return f"subtests {outcome}", "u", f"SUBPASSED{description}"
  323. elif report.skipped:
  324. if quiet:
  325. return "", "", ""
  326. else:
  327. return outcome, "-", f"SUBSKIPPED{description}"
  328. else:
  329. failed_subtests_count = config.stash[failed_subtests_key][report.nodeid]
  330. # Top-level test, fail if it contains failed subtests and it has passed.
  331. if report.passed and failed_subtests_count > 0:
  332. report.outcome = "failed"
  333. suffix = "s" if failed_subtests_count > 1 else ""
  334. report.longrepr = f"contains {failed_subtests_count} failed subtest{suffix}"
  335. return None