feat!: Removed event_loop fixture.

Author: seifertm

docs/reference/fixtures/event_loop_example.py

@@ -2,33 +2,6 @@
 Fixtures
 ========
 
-event_loop
-==========
-*This fixture is deprecated.*
-
-*If you want to request an asyncio event loop with a scope other than function
-scope, use the "loop_scope" argument to* :ref:`reference/markers/asyncio` *when marking the tests.
-If you want to return different types of event loops, use the* :ref:`reference/fixtures/event_loop_policy`
-*fixture.*
-
-Creates a new asyncio event loop based on the current event loop policy. The new loop
-is available as the return value of this fixture for synchronous functions, or via `asyncio.get_running_loop <https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.get_running_loop>`__ for asynchronous functions.
-The event loop is closed when the fixture scope ends.
-The fixture scope defaults to ``function`` scope.
-
-.. include:: event_loop_example.py
- :code: python
-
-Note that, when using the ``event_loop`` fixture, you need to interact with the event loop using methods like ``event_loop.run_until_complete``. If you want to *await* code inside your test function, you need to write a coroutine and use it as a test function. The :ref:`asyncio <reference/markers/asyncio>` marker
-is used to mark coroutines that should be treated as test functions.
-
-If you need to change the type of the event loop, prefer setting a custom event loop policy over redefining the ``event_loop`` fixture.
-
-If the ``pytest.mark.asyncio`` decorator is applied to a test function, the ``event_loop``
-fixture will be requested automatically by the test function.
-
-.. _reference/fixtures/event_loop_policy:
-
 event_loop_policy
 =================
 Returns the event loop policy used to create asyncio event loops.

docs/reference/fixtures/index.rst

@@ -259,14 +259,17 @@ def _preprocess_async_fixtures(
  # Ignore async fixtures without explicit asyncio mark in strict mode
  # This applies to pytest_trio fixtures, for example
  continue
- scope = (
+ loop_scope = (
  getattr(func, "_loop_scope", None)
  or default_loop_scope
  or fixturedef.scope
  )
- if scope == "function" and "event_loop" not in fixturedef.argnames:
- fixturedef.argnames += ("event_loop",)
- _make_asyncio_fixture_function(func, scope)
+ if (
+ loop_scope == "function"
+ and "_function_event_loop" not in fixturedef.argnames
+ ):
+ fixturedef.argnames += ("_function_event_loop",)
+ _make_asyncio_fixture_function(func, loop_scope)
  function_signature = inspect.signature(func)
  if "event_loop" in function_signature.parameters:
  warnings.warn(
@@ -415,7 +418,7 @@ def _get_event_loop_fixture_id_for_async_fixture(
  getattr(func, "_loop_scope", None) or default_loop_scope or request.scope
  )
  if loop_scope == "function":
- event_loop_fixture_id = "event_loop"
+ event_loop_fixture_id = "_function_event_loop"
  else:
  event_loop_node = _retrieve_scope_root(request._pyfuncitem, loop_scope)
  event_loop_fixture_id = event_loop_node.stash.get(
@@ -1101,7 +1104,7 @@ def pytest_runtest_setup(item: pytest.Item) -> None:
  parent_node = _retrieve_scope_root(item, scope)
  event_loop_fixture_id = parent_node.stash[_event_loop_fixture_id]
  else:
- event_loop_fixture_id = "event_loop"
+ event_loop_fixture_id = "_function_event_loop"
  fixturenames = item.fixturenames # type: ignore[attr-defined]
  if event_loop_fixture_id not in fixturenames:
  fixturenames.append(event_loop_fixture_id)
@@ -1170,11 +1173,20 @@ def _retrieve_scope_root(item: Collector | Item, scope: str) -> Collector:
  raise pytest.UsageError(error_message)
 
 
-@pytest.fixture
-def event_loop(request: FixtureRequest) -> Iterator[asyncio.AbstractEventLoop]:
- """Create an instance of the default event loop for each test case."""
- new_loop_policy = request.getfixturevalue(event_loop_policy.__name__)
- with _temporary_event_loop_policy(new_loop_policy), _provide_event_loop() as loop:
+@pytest.fixture(
+ scope="function",
+ name="_function_event_loop",
+)
+def _function_event_loop(
+ *args, # Function needs to accept "cls" when collected by pytest.Class
+ event_loop_policy,
+) -> Iterator[asyncio.AbstractEventLoop]:
+ new_loop_policy = event_loop_policy
+ with (
+ _temporary_event_loop_policy(new_loop_policy),
+ _provide_event_loop() as loop,
+ ):
+ asyncio.set_event_loop(loop)
  yield loop
 
 

pytest_asyncio/plugin.py

@@ -9,6 +9,8 @@
 
 import pytest
 
+import pytest_asyncio
+
 
 @pytest.fixture(scope="module")
 def event_loop():
@@ -18,7 +20,7 @@ def event_loop():
  loop.close()
 
 
-@pytest.fixture(scope="module")
+@pytest_asyncio.fixture(scope="module", loop_scope="module")
 async def async_fixture():
  await asyncio.sleep(0.1)
  return 1

tests/async_fixtures/test_async_fixtures_scope.py

@@ -1,34 +1,3 @@
 from __future__ import annotations
 
-import asyncio
-
-import pytest
-
 pytest_plugins = "pytester"
-
-
-@pytest.fixture
-def dependent_fixture(event_loop):
- """A fixture dependent on the event_loop fixture, doing some cleanup."""
- counter = 0
-
- async def just_a_sleep():
- """Just sleep a little while."""
- nonlocal event_loop
- await asyncio.sleep(0.1)
- nonlocal counter
- counter += 1
-
- event_loop.run_until_complete(just_a_sleep())
- yield
- event_loop.run_until_complete(just_a_sleep())
-
- assert counter == 2
-
-
-@pytest.fixture(scope="session", name="factory_involving_factories")
-def factory_involving_factories_fixture(unused_tcp_port_factory):
- def factory():
- return unused_tcp_port_factory()
-
- return factory

tests/async_fixtures/test_parametrized_loop.py

@@ -45,43 +45,6 @@ async def test_mark_and_parametrize(x, y):
  assert y in (1, 2)
 
 
-def test_can_use_explicit_event_loop_fixture(pytester: Pytester):
- pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = module")
- pytester.makepyfile(
- dedent(
- """\
- import asyncio
- import pytest
- from hypothesis import given
- import hypothesis.strategies as st
-
- pytest_plugins = 'pytest_asyncio'
-
- @pytest.fixture(scope="module")
- def event_loop():
- loop = asyncio.get_event_loop_policy().new_event_loop()
- yield loop
- loop.close()
-
- @given(st.integers())
- @pytest.mark.asyncio
- async def test_explicit_fixture_request(event_loop, n):
- semaphore = asyncio.Semaphore(value=0)
- event_loop.call_soon(semaphore.release)
- await semaphore.acquire()
- """
- )
- )
- result = pytester.runpytest_subprocess("--asyncio-mode=strict", "-W default")
- result.assert_outcomes(passed=1, warnings=2)
- result.stdout.fnmatch_lines(
- [
- '*is asynchronous and explicitly requests the "event_loop" fixture*',
- "*event_loop fixture provided by pytest-asyncio has been redefined*",
- ]
- )
-
-
 def test_async_auto_marked(pytester: Pytester):
  pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
  pytester.makepyfile(