Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

use Contextmanagers to handle StopIteration in generators #12934

Draft
wants to merge 7 commits into
base: main
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions src/_pytest/logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -809,6 +809,7 @@ def pytest_runtest_logstart(self) -> None:
def pytest_runtest_logreport(self) -> None:
self.log_cli_handler.set_when("logreport")

@contextmanager
def _runtest_for(self, item: nodes.Item, when: str) -> Generator[None]:
"""Implement the internals of the pytest_runtest_xxx() hooks."""
with (
Expand Down Expand Up @@ -838,20 +839,23 @@ def pytest_runtest_setup(self, item: nodes.Item) -> Generator[None]:

empty: dict[str, list[logging.LogRecord]] = {}
item.stash[caplog_records_key] = empty
yield from self._runtest_for(item, "setup")
with self._runtest_for(item, "setup"):
yield

@hookimpl(wrapper=True)
def pytest_runtest_call(self, item: nodes.Item) -> Generator[None]:
self.log_cli_handler.set_when("call")

yield from self._runtest_for(item, "call")
with self._runtest_for(item, "call"):
yield

@hookimpl(wrapper=True)
def pytest_runtest_teardown(self, item: nodes.Item) -> Generator[None]:
self.log_cli_handler.set_when("teardown")

try:
yield from self._runtest_for(item, "teardown")
with self._runtest_for(item, "teardown"):
yield
finally:
del item.stash[caplog_records_key]
del item.stash[caplog_handler_key]
Expand Down
50 changes: 50 additions & 0 deletions testing/acceptance_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -1600,3 +1600,53 @@ def test_no_terminal_plugin(pytester: Pytester) -> None:
pytester.makepyfile("def test(): assert 1 == 2")
result = pytester.runpytest("-pno:terminal", "-s")
assert result.ret == ExitCode.TESTS_FAILED


def test_stop_iteration_from_collect(pytester: Pytester) -> None:
pytester.makepyfile(test_it="raise StopIteration('hello')")
result = pytester.runpytest()
assert result.ret == ExitCode.INTERRUPTED
result.assert_outcomes(failed=0, passed=0, errors=1)
result.stdout.fnmatch_lines(
[
"=========================== short test summary info ============================",
"ERROR test_it.py - StopIteration: hello",
"!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!",
"=============================== 1 error in * ===============================",
]
)


def test_stop_iteration_runtest_protocol(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
import pytest

@pytest.fixture
def fail_setup():
raise StopIteration(1)

def test_fail_setup(fail_setup):
pass

def test_fail_teardown(request):
def stop_iteration():
raise StopIteration(2)
request.addfinalizer(stop_iteration)
graingert marked this conversation as resolved.
Show resolved Hide resolved

def test_fail_call():
raise StopIteration(3)
"""
)
result = pytester.runpytest()
assert result.ret == ExitCode.TESTS_FAILED
result.assert_outcomes(failed=1, passed=1, errors=2)
result.stdout.fnmatch_lines(
[
"=========================== short test summary info ============================",
"FAILED test_it.py::test_fail_call - StopIteration: 3",
"ERROR test_it.py::test_fail_setup - StopIteration: 1",
"ERROR test_it.py::test_fail_teardown - StopIteration: 2",
"==================== 1 failed, 1 passed, 2 errors in * =====================",
]
)
10 changes: 10 additions & 0 deletions testing/example_scripts/hook_exceptions/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
from __future__ import annotations

from collections.abc import Iterator

import pytest


@pytest.hookimpl(wrapper=True)
def pytest_runtest_call() -> Iterator[None]:
yield
Empty file.
87 changes: 87 additions & 0 deletions testing/example_scripts/hook_exceptions/test_stop_iteration.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
"""
test example file exposing mltiple issues with corutine exception passover in case of stopiteration
Copy link
Member

@graingert graingert Dec 19, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
test example file exposing mltiple issues with corutine exception passover in case of stopiteration
test example file exposing multiple issues with coroutine exception passover in case of stopiteration


the stdlib contextmanager implementation explicitly catches
and reshapes in case a StopIteration was send in and is raised out
"""

from __future__ import annotations

from collections.abc import Iterator
from contextlib import contextmanager

import pluggy


def test_stop() -> None:
raise StopIteration()


hookspec = pluggy.HookspecMarker("myproject")
hookimpl = pluggy.HookimplMarker("myproject")


class MySpec:
"""A hook specification namespace."""

@hookspec
def myhook(self, arg1: int, arg2: int) -> int: # type: ignore[empty-body]
"""My special little hook that you can customize."""


class Plugin_1:
"""A hook implementation namespace."""

@hookimpl
def myhook(self, arg1: int, arg2: int) -> int:
print("inside Plugin_1.myhook()")
raise StopIteration()


class Plugin_2:
"""A 2nd hook implementation namespace."""

@hookimpl(wrapper=True)
def myhook(self) -> Iterator[None]:
return (yield)


def try_pluggy() -> None:
# create a manager and add the spec
pm = pluggy.PluginManager("myproject")
pm.add_hookspecs(MySpec)

# register plugins
pm.register(Plugin_1())
pm.register(Plugin_2())

# call our ``myhook`` hook
results = pm.hook.myhook(arg1=1, arg2=2)
print(results)


@contextmanager
def my_cm() -> Iterator[None]:
try:
yield
except Exception as e:
print(e)
raise StopIteration()


def inner() -> None:
with my_cm():
raise StopIteration()


def try_context() -> None:
inner()


mains = {"pluggy": try_pluggy, "context": try_context}

if __name__ == "__main__":
import sys

if len(sys.argv) == 2:
mains[sys.argv[1]]()
Loading