Support -x/--exitfirst

In order to support that and provide a good error message, we needed to stop using @contextmanager so we could have full control over the error stack.

Fix #23

Author: rhoban13

CHANGELOG.rst

@@ -4,9 +4,11 @@ CHANGELOG
 UNRELEASED
 ----------
 
+* Support ``-x/--exitfirst`` (`#134`_).
 * Hide the traceback inside the ``SubTests.test()`` method (`#131`_).
 
 .. _#131: https://github.com/pytest-dev/pytest-subtests/pull/131
+.. _#134: https://github.com/pytest-dev/pytest-subtests/pull/134
 
 0.12.1 (2024-03-07)
 -------------------

src/pytest_subtests/plugin.py

@@ -3,6 +3,7 @@
 import sys
 import time
 from contextlib import contextmanager
+from contextlib import ExitStack
 from contextlib import nullcontext
 from typing import Any
 from typing import Callable
@@ -218,47 +219,109 @@ def _capturing_logs(self) -> Generator[CapturedLogs | NullCapturedLogs, None, No
             with catching_logs(handler):
                 yield captured_logs
 
-    @contextmanager
     def test(
         self,
         msg: str | None = None,
         **kwargs: Any,
-    ) -> Generator[None, None, None]:
-        # Hide from tracebacks.
+    ) -> _SubTestContextManager:
+        """
+        Context manager for subtests, capturing exceptions raised inside the subtest scope and handling
+        them through the pytest machinery.
+
+        Usage:
+
+        .. code-block:: python
+
+            with subtests.test(msg="subtest"):
+                assert 1 == 1
+        """
+        return _SubTestContextManager(
+            self.ihook,
+            msg,
+            kwargs,
+            capturing_output_ctx=self._capturing_output,
+            capturing_logs_ctx=self._capturing_logs,
+            request=self.request,
+            suspend_capture_ctx=self.suspend_capture_ctx,
+        )
+
+
+@attr.s(auto_attribs=True)
+class _SubTestContextManager:
+    """
+    Context manager for subtests, capturing exceptions raised inside the subtest scope and handling
+    them through the pytest machinery.
+
+    Note: initially this logic was implemented directly in SubTests.test() as a @contextmanager, however
+    it is not possible to control the output fully when exiting from it due to an exception when
+    in --exitfirst mode, so this was refactored into an explicit context manager class (#134).
+    """
+
+    ihook: pluggy.HookRelay
+    msg: str | None
+    kwargs: dict[str, Any]
+    capturing_output_ctx: Callable[[], ContextManager]
+    capturing_logs_ctx: Callable[[], ContextManager]
+    suspend_capture_ctx: Callable[[], ContextManager]
+    request: SubRequest
+
+    def __enter__(self) -> None:
         __tracebackhide__ = True
 
-        start = time.time()
-        precise_start = time.perf_counter()
-        exc_info = None
+        self._start = time.time()
+        self._precise_start = time.perf_counter()
+        self._exc_info = None
+
+        self._exit_stack = ExitStack()
+        self._captured_output = self._exit_stack.enter_context(
+            self.capturing_output_ctx()
+        )
+        self._captured_logs = self._exit_stack.enter_context(self.capturing_logs_ctx())
+
+    def __exit__(
+        self,
+        exc_type: type[Exception] | None,
+        exc_val: Exception | None,
+        exc_tb: TracebackType | None,
+    ) -> bool:
+        __tracebackhide__ = True
+        try:
+            if exc_val is not None:
+                if self.request.session.shouldfail:
+                    return False
 
-        with self._capturing_output() as captured_output, self._capturing_logs() as captured_logs:
-            try:
-                yield
-            except (Exception, OutcomeException):
-                exc_info = ExceptionInfo.from_current()
+                exc_info = ExceptionInfo.from_exception(exc_val)
+            else:
+                exc_info = None
+        finally:
+            self._exit_stack.close()
 
         precise_stop = time.perf_counter()
-        duration = precise_stop - precise_start
+        duration = precise_stop - self._precise_start
         stop = time.time()
 
         call_info = make_call_info(
-            exc_info, start=start, stop=stop, duration=duration, when="call"
+            exc_info, start=self._start, stop=stop, duration=duration, when="call"
+        )
+        report = self.ihook.pytest_runtest_makereport(
+            item=self.request.node, call=call_info
         )
-        report = self.ihook.pytest_runtest_makereport(item=self.item, call=call_info)
         sub_report = SubTestReport._from_test_report(report)
-        sub_report.context = SubTestContext(msg, kwargs.copy())
+        sub_report.context = SubTestContext(self.msg, self.kwargs.copy())
 
-        captured_output.update_report(sub_report)
-        captured_logs.update_report(sub_report)
+        self._captured_output.update_report(sub_report)
+        self._captured_logs.update_report(sub_report)
 
         with self.suspend_capture_ctx():
             self.ihook.pytest_runtest_logreport(report=sub_report)
 
         if check_interactive_exception(call_info, sub_report):
             self.ihook.pytest_exception_interact(
-                node=self.item, call=call_info, report=sub_report
+                node=self.request.node, call=call_info, report=sub_report
             )
 
+        return True
+
 
 def make_call_info(
     exc_info: ExceptionInfo[BaseException] | None,

tests/test_subtests.py

@@ -580,3 +580,29 @@ def runpytest_and_check_pdb(
         # assert.
         result.stdout.fnmatch_lines("*entering PDB*")
         assert self._FakePdb.calls == ["init", "reset", "interaction"]
+
+
+def test_exitfirst(pytester: pytest.Pytester) -> None:
+    """
+    Validate that when passing --exitfirst the test exits after the first failed subtest.
+    """
+    pytester.makepyfile(
+        """
+        def test_foo(subtests):
+            with subtests.test("sub1"):
+                assert False
+
+            with subtests.test("sub2"):
+                pass
+        """
+    )
+    result = pytester.runpytest("--exitfirst")
+    assert result.parseoutcomes()["failed"] == 1
+    result.stdout.fnmatch_lines(
+        [
+            "*[[]sub1[]] SUBFAIL test_exitfirst.py::test_foo - assert False*",
+            "* stopping after 1 failures*",
+        ],
+        consecutive=True,
+    )
+    result.stdout.no_fnmatch_line("*sub2*")  # sub2 not executed.