Skip to content

Scoped event loops based on pytest marks #620

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

Merged
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
Prev Previous commit
Next Next commit
[feat] The asyncio_event_loop mark provides a module-scoped asyncio e…
…vent loop when a module has the mark.

Signed-off-by: Michael Seifert <m.seifert@digitalernachschub.de>
  • Loading branch information
seifertm committed Oct 9, 2023
commit 0c92628826d1a867361afbca5b3b0ee42100e2eb
29 changes: 27 additions & 2 deletions docs/source/reference/markers.rst
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,10 @@ automatically to *async* test functions.

``pytest.mark.asyncio_event_loop``
==================================
Test classes with this mark provide a class-scoped asyncio event loop.
Test classes or modules with this mark provide a class-scoped or module-scoped asyncio event loop.

This functionality is orthogonal to the `asyncio` mark.
That means the presence of this mark does not imply that async test functions inside the class are collected by pytest-asyncio.
That means the presence of this mark does not imply that async test functions inside the class or module are collected by pytest-asyncio.
The collection happens automatically in `auto` mode.
However, if you're using strict mode, you still have to apply the `asyncio` mark to your async test functions.

Expand Down Expand Up @@ -79,8 +79,33 @@ In *auto* mode, the ``pytest.mark.asyncio`` marker can be omitted:
async def test_this_runs_in_same_loop(self):
assert asyncio.get_running_loop() is TestClassScopedLoop.loop

Similarly, a module-scoped loop is provided when adding the `asyncio_event_loop` mark to the module:

.. code-block:: python

import asyncio

import pytest

pytestmark = pytest.mark.asyncio_event_loop

loop: asyncio.AbstractEventLoop


async def test_remember_loop():
global loop
loop = asyncio.get_running_loop()


async def test_this_runs_in_same_loop():
global loop
assert asyncio.get_running_loop() is loop


class TestClassA:
async def test_this_runs_in_same_loop(self):
global loop
assert asyncio.get_running_loop() is loop

.. |pytestmark| replace:: ``pytestmark``
.. _pytestmark: http://doc.pytest.org/en/latest/example/markers.html#marking-whole-classes-or-modules
12 changes: 8 additions & 4 deletions pytest_asyncio/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,8 @@ def pytest_configure(config: Config) -> None:
config.addinivalue_line(
"markers",
"asyncio_event_loop: "
"Provides an asyncio event loop in the scope of the marked test class",
"Provides an asyncio event loop in the scope of the marked test "
"class or module",
)


Expand Down Expand Up @@ -347,7 +348,7 @@ def pytest_pycollect_makeitem(

@pytest.hookimpl
def pytest_collectstart(collector: pytest.Collector):
if not isinstance(collector, pytest.Class):
if not isinstance(collector, (pytest.Class, pytest.Module)):
return
# pytest.Collector.own_markers is empty at this point,
# so we rely on _pytest.mark.structures.get_unpacked_marks
Expand All @@ -357,17 +358,20 @@ def pytest_collectstart(collector: pytest.Collector):
continue

@pytest.fixture(
scope="class",
scope="class" if isinstance(collector, pytest.Class) else "module",
name="event_loop",
)
def scoped_event_loop(cls) -> Iterator[asyncio.AbstractEventLoop]:
def scoped_event_loop(
*args, # Function needs to accept "cls" when collected by pytest.Class
) -> Iterator[asyncio.AbstractEventLoop]:
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()

# @pytest.fixture does not register the fixture anywhere, so pytest doesn't
# know it exists. We work around this by attaching the fixture function to the
# collected Python class, where it will be picked up by pytest.Class.collect()
# or pytest.Module.collect(), respectively
collector.obj.__pytest_asyncio_scoped_event_loop = scoped_event_loop
break

Expand Down
63 changes: 63 additions & 0 deletions tests/markers/test_module_marker.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,3 +50,66 @@ def sample_fixture():
)
result = pytester.runpytest("--asyncio-mode=strict")
result.assert_outcomes(passed=2)


def test_asyncio_mark_provides_module_scoped_loop_strict_mode(pytester: Pytester):
pytester.makepyfile(
dedent(
"""\
import asyncio
import pytest

pytestmark = pytest.mark.asyncio_event_loop

loop: asyncio.AbstractEventLoop

@pytest.mark.asyncio
async def test_remember_loop():
global loop
loop = asyncio.get_running_loop()

@pytest.mark.asyncio
async def test_this_runs_in_same_loop():
global loop
assert asyncio.get_running_loop() is loop

class TestClassA:
@pytest.mark.asyncio
async def test_this_runs_in_same_loop(self):
global loop
assert asyncio.get_running_loop() is loop
"""
)
)
result = pytester.runpytest("--asyncio-mode=strict")
result.assert_outcomes(passed=3)


def test_asyncio_mark_provides_class_scoped_loop_auto_mode(pytester: Pytester):
pytester.makepyfile(
dedent(
"""\
import asyncio
import pytest

pytestmark = pytest.mark.asyncio_event_loop

loop: asyncio.AbstractEventLoop

async def test_remember_loop():
global loop
loop = asyncio.get_running_loop()

async def test_this_runs_in_same_loop():
global loop
assert asyncio.get_running_loop() is loop

class TestClassA:
async def test_this_runs_in_same_loop(self):
global loop
assert asyncio.get_running_loop() is loop
"""
)
)
result = pytester.runpytest("--asyncio-mode=auto")
result.assert_outcomes(passed=3)