Skip to content
Open
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
1 change: 1 addition & 0 deletions changelog.d/1517.removed.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Deprecated *event_loop_policy* fixture
1 change: 0 additions & 1 deletion docs/how-to-guides/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ How-To Guides
run_class_tests_in_same_loop
run_module_tests_in_same_loop
run_package_tests_in_same_loop
multiple_loops
parametrize_with_asyncio
uvloop
test_item_is_async
Expand Down
14 changes: 0 additions & 14 deletions docs/how-to-guides/multiple_loops.rst

This file was deleted.

29 changes: 0 additions & 29 deletions docs/how-to-guides/multiple_loops_example.py

This file was deleted.

21 changes: 0 additions & 21 deletions docs/how-to-guides/uvloop.rst
Original file line number Diff line number Diff line change
Expand Up @@ -18,24 +18,3 @@ Define a ``pytest_asyncio_loop_factories`` hook in your *conftest.py* that maps

:doc:`custom_loop_factory`
More details on the ``pytest_asyncio_loop_factories`` hook, including per-test factory selection and multiple factory parametrization.

Using the event_loop_policy fixture
-----------------------------------

.. note::

``asyncio.AbstractEventLoopPolicy`` is deprecated as of Python 3.14 (removal planned for 3.16), and ``uvloop.EventLoopPolicy`` will be removed alongside it. Overriding the *event_loop_policy* fixture is also deprecated in pytest-asyncio. Prefer the hook approach above.

For older versions of Python and uvloop, you can override the *event_loop_policy* fixture in your *conftest.py:*

.. code-block:: python

import pytest
import uvloop


@pytest.fixture(scope="session")
def event_loop_policy():
return uvloop.EventLoopPolicy()

You may choose to limit the scope of the fixture to *package,* *module,* or *class,* if you only want a subset of your tests to run with uvloop.
23 changes: 0 additions & 23 deletions docs/reference/fixtures/event_loop_policy_example.py

This file was deleted.

This file was deleted.

23 changes: 0 additions & 23 deletions docs/reference/fixtures/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -2,29 +2,6 @@
Fixtures
========

event_loop_policy
=================

.. warning::

Overriding the *event_loop_policy* fixture is deprecated and will be removed in a future version of pytest-asyncio. Use the ``pytest_asyncio_loop_factories`` hook instead. See :doc:`../hooks` for details.

Returns the event loop policy used to create asyncio event loops.
The default return value is *asyncio.get_event_loop_policy().*

This fixture can be overridden when a different event loop policy should be used.

.. include:: event_loop_policy_example.py
:code: python

Multiple policies can be provided via fixture parameters.
The fixture is automatically applied to all pytest-asyncio tests.
Therefore, all tests managed by pytest-asyncio are run once for each fixture parameter.
The following example runs the test with different event loop policies.

.. include:: event_loop_policy_parametrized_example.py
:code: python

unused_tcp_port
===============
Finds and yields a single unused TCP port on the localhost interface. Useful for
Expand Down
28 changes: 2 additions & 26 deletions pytest_asyncio/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -805,11 +805,8 @@ def _temporary_event_loop(loop: AbstractEventLoop) -> Iterator[None]:


@contextlib.contextmanager
def _temporary_event_loop_policy(
policy: AbstractEventLoopPolicy,
) -> Iterator[None]:
def _restore_event_loop_policy() -> Iterator[None]:
old_loop_policy = _get_event_loop_policy()
_set_event_loop_policy(policy)
try:
yield
finally:
Expand Down Expand Up @@ -910,13 +907,6 @@ def inner(*args, **kwargs):

@pytest.hookimpl(wrapper=True)
def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None:
if (
fixturedef.argname == "event_loop_policy"
and fixturedef.func.__module__ != __name__
):
warnings.warn(
PytestDeprecationWarning(_EVENT_LOOP_POLICY_FIXTURE_DEPRECATION_WARNING),
)
asyncio_mode = _get_asyncio_mode(request.config)
if not _is_asyncio_fixture_function(fixturedef.func):
if asyncio_mode == Mode.STRICT:
Expand Down Expand Up @@ -960,12 +950,6 @@ def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None:
mark.asyncio 'loop_factories' must be a non-empty sequence of strings.
"""

_EVENT_LOOP_POLICY_FIXTURE_DEPRECATION_WARNING = """\
Overriding the "event_loop_policy" fixture is deprecated \
and will be removed in a future version of pytest-asyncio. \
Use the "pytest_asyncio_loop_factories" hook to customize event loop creation.\
"""


def _parse_asyncio_marker(
asyncio_marker: Mark,
Expand Down Expand Up @@ -1027,13 +1011,11 @@ def _create_scoped_runner_fixture(scope: _ScopeName) -> Callable:
name=f"_{scope}_scoped_runner",
)
def _scoped_runner(
event_loop_policy,
_asyncio_loop_factory,
request: FixtureRequest,
) -> Iterator[Runner]:
new_loop_policy = event_loop_policy
debug_mode = _get_asyncio_debug(request.config)
with _temporary_event_loop_policy(new_loop_policy):
with _restore_event_loop_policy():
runner = Runner(
debug=debug_mode,
loop_factory=_asyncio_loop_factory,
Expand Down Expand Up @@ -1074,12 +1056,6 @@ def _asyncio_loop_factory(request: FixtureRequest) -> LoopFactory | None:
return getattr(request, "param", None)


@pytest.fixture(scope="session", autouse=True)
def event_loop_policy() -> AbstractEventLoopPolicy:
"""Return an instance of the policy used to create asyncio event loops."""
return _get_event_loop_policy()


def is_async_test(item: Item) -> TypeIs[PytestAsyncioFunction]:
"""Returns whether a test item is a pytest-asyncio test"""
return isinstance(item, PytestAsyncioFunction)
Expand Down
78 changes: 0 additions & 78 deletions tests/markers/test_class_scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
from __future__ import annotations

import asyncio
import sys
from textwrap import dedent

import pytest
Expand Down Expand Up @@ -98,83 +97,6 @@ async def test_this_runs_in_same_loop(self):
result.assert_outcomes(passed=2)


def test_asyncio_mark_respects_the_loop_policy(
pytester: pytest.Pytester,
):
pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
pytester.makepyfile(dedent("""\
import asyncio
import pytest

class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
pass

class TestUsesCustomEventLoop:
@pytest.fixture(scope="class")
@classmethod
def event_loop_policy(cls):
return CustomEventLoopPolicy()

@pytest.mark.asyncio
async def test_uses_custom_event_loop_policy(self):
assert isinstance(
asyncio.get_event_loop_policy(),
CustomEventLoopPolicy,
)

@pytest.mark.asyncio
async def test_does_not_use_custom_event_loop_policy():
assert not isinstance(
asyncio.get_event_loop_policy(),
CustomEventLoopPolicy,
)
"""))
pytest_args = ["--asyncio-mode=strict"]
if sys.version_info >= (3, 14):
pytest_args.extend(["-W", "default"])
result = pytester.runpytest(*pytest_args)
if sys.version_info >= (3, 14):
result.assert_outcomes(passed=2, warnings=4)
result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*")
else:
result.assert_outcomes(passed=2)


def test_asyncio_mark_respects_parametrized_loop_policies(
pytester: pytest.Pytester,
):
pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
pytester.makepyfile(dedent("""\
import asyncio

import pytest

@pytest.fixture(
scope="class",
params=[
asyncio.DefaultEventLoopPolicy(),
asyncio.DefaultEventLoopPolicy(),
]
)
def event_loop_policy(request):
return request.param

@pytest.mark.asyncio(loop_scope="class")
class TestWithDifferentLoopPolicies:
async def test_parametrized_loop(self, request):
pass
"""))
pytest_args = ["--asyncio-mode=strict"]
if sys.version_info >= (3, 14):
pytest_args.extend(["-W", "default"])
result = pytester.runpytest(*pytest_args)
if sys.version_info >= (3, 14):
result.assert_outcomes(passed=2, warnings=4)
result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*")
else:
result.assert_outcomes(passed=2)


def test_asyncio_mark_provides_class_scoped_loop_to_fixtures(
pytester: pytest.Pytester,
):
Expand Down
Loading
Loading