remove use of deprecated policy API from tests (#10851)

Author: kumaraditya303

CHANGES/10851.bugfix.rst

@@ -0,0 +1 @@
+Fixed pytest plugin to not use deprecated :py:mod:`asyncio` policy APIs.

CHANGES/10851.contrib.rst

@@ -0,0 +1,2 @@
+Updated tests to avoid using deprecated :py:mod:`asyncio` policy APIs and
+make it compatible with Python 3.14.

aiohttp/pytest_plugin.py

@@ -224,9 +224,13 @@ def pytest_pyfunc_call(pyfuncitem):  # type: ignore[no-untyped-def]
     """Run coroutines in an event loop instead of a normal function call."""
     fast = pyfuncitem.config.getoption("--aiohttp-fast")
     if inspect.iscoroutinefunction(pyfuncitem.function):
-        existing_loop = pyfuncitem.funcargs.get(
-            "proactor_loop"
-        ) or pyfuncitem.funcargs.get("loop", None)
+        existing_loop = (
+            pyfuncitem.funcargs.get("proactor_loop")
+            or pyfuncitem.funcargs.get("selector_loop")
+            or pyfuncitem.funcargs.get("uvloop_loop")
+            or pyfuncitem.funcargs.get("loop", None)
+        )
+
         with _runtime_warning_context():
             with _passthrough_loop_context(existing_loop, fast=fast) as _loop:
                 testargs = {
@@ -243,11 +247,11 @@ def pytest_generate_tests(metafunc):  # type: ignore[no-untyped-def]
         return
 
     loops = metafunc.config.option.aiohttp_loop
-    avail_factories: Dict[str, Type[asyncio.AbstractEventLoopPolicy]]
-    avail_factories = {"pyloop": asyncio.DefaultEventLoopPolicy}
+    avail_factories: dict[str, Callable[[], asyncio.AbstractEventLoop]]
+    avail_factories = {"pyloop": asyncio.new_event_loop}
 
     if uvloop is not None:
-        avail_factories["uvloop"] = uvloop.EventLoopPolicy
+        avail_factories["uvloop"] = uvloop.new_event_loop
 
     if loops == "all":
         loops = "pyloop,uvloop?"
@@ -272,14 +276,12 @@ def pytest_generate_tests(metafunc):  # type: ignore[no-untyped-def]
 
 @pytest.fixture
 def loop(
-    loop_factory: Callable[[], asyncio.AbstractEventLoopPolicy],
+    loop_factory: Callable[[], asyncio.AbstractEventLoop],
     fast: bool,
     loop_debug: bool,
 ) -> Iterator[asyncio.AbstractEventLoop]:
     """Return an instance of the event loop."""
-    policy = loop_factory()
-    asyncio.set_event_loop_policy(policy)
-    with loop_context(fast=fast) as _loop:
+    with loop_context(loop_factory, fast=fast) as _loop:
         if loop_debug:
             _loop.set_debug(True)
         asyncio.set_event_loop(_loop)
@@ -288,10 +290,9 @@ def loop(
 
 @pytest.fixture
 def proactor_loop() -> Iterator[asyncio.AbstractEventLoop]:
-    policy = asyncio.WindowsProactorEventLoopPolicy()  # type: ignore[attr-defined]
-    asyncio.set_event_loop_policy(policy)
+    factory = asyncio.ProactorEventLoop  # type: ignore[attr-defined]
 
-    with loop_context(policy.new_event_loop) as _loop:
+    with loop_context(factory) as _loop:
         asyncio.set_event_loop(_loop)
         yield _loop
 

tests/conftest.py

@@ -233,20 +233,16 @@ def assert_sock_fits(sock_path: str) -> None:
 
 @pytest.fixture
 def selector_loop() -> Iterator[asyncio.AbstractEventLoop]:
-    policy = asyncio.WindowsSelectorEventLoopPolicy()  # type: ignore[attr-defined]
-    asyncio.set_event_loop_policy(policy)
-
-    with loop_context(policy.new_event_loop) as _loop:
+    factory = asyncio.SelectorEventLoop
+    with loop_context(factory) as _loop:
         asyncio.set_event_loop(_loop)
         yield _loop
 
 
 @pytest.fixture
 def uvloop_loop() -> Iterator[asyncio.AbstractEventLoop]:
-    policy = uvloop.EventLoopPolicy()
-    asyncio.set_event_loop_policy(policy)
-
-    with loop_context(policy.new_event_loop) as _loop:
+    factory = uvloop.new_event_loop
+    with loop_context(factory) as _loop:
         asyncio.set_event_loop(_loop)
         yield _loop
 

tests/test_connector.py

@@ -124,7 +124,7 @@ def create_mocked_conn(
     try:
         loop = asyncio.get_running_loop()
     except RuntimeError:
-        loop = asyncio.get_event_loop_policy().get_event_loop()
+        loop = asyncio.get_event_loop()
 
     f = loop.create_future()
     proto: mock.Mock = mock.create_autospec(

tests/test_loop.py

@@ -37,7 +37,7 @@ async def test_on_startup_hook(self) -> None:
 
 
 def test_default_loop(loop: asyncio.AbstractEventLoop) -> None:
-    assert asyncio.get_event_loop_policy().get_event_loop() is loop
+    assert asyncio.get_event_loop() is loop
 
 
 def test_setup_loop_non_main_thread() -> None:
@@ -46,7 +46,7 @@ def test_setup_loop_non_main_thread() -> None:
     def target() -> None:
         try:
             with loop_context() as loop:
-                assert asyncio.get_event_loop_policy().get_event_loop() is loop
+                assert asyncio.get_event_loop() is loop
                 loop.run_until_complete(test_subprocess_co(loop))
         except Exception as exc:
             nonlocal child_exc

tests/test_proxy_functional.py

@@ -241,7 +241,6 @@ async def test_https_proxy_unsupported_tls_in_tls(
     await asyncio.sleep(0.1)
 
 
-@pytest.mark.usefixtures("uvloop_loop")
 @pytest.mark.skipif(
     platform.system() == "Windows" or sys.implementation.name != "cpython",
     reason="uvloop is not supported on Windows and non-CPython implementations",
@@ -253,6 +252,7 @@ async def test_https_proxy_unsupported_tls_in_tls(
 async def test_uvloop_secure_https_proxy(
     client_ssl_ctx: ssl.SSLContext,
     secure_proxy_url: URL,
+    uvloop_loop: asyncio.AbstractEventLoop,
 ) -> None:
     """Ensure HTTPS sites are accessible through a secure proxy without warning when using uvloop."""
     conn = aiohttp.TCPConnector()