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

(cherry picked from commit e5d1240)

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

@@ -10,7 +10,6 @@
     Iterator,
     Optional,
     Protocol,
-    Type,
     Union,
     overload,
 )
@@ -208,9 +207,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 = {
@@ -227,11 +230,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:  # pragma: no cover
-        avail_factories["uvloop"] = uvloop.EventLoopPolicy
+        avail_factories["uvloop"] = uvloop.new_event_loop
 
     if loops == "all":
         loops = "pyloop,uvloop?"
@@ -255,23 +258,24 @@ def pytest_generate_tests(metafunc):  # type: ignore[no-untyped-def]
 
 
 @pytest.fixture
-def loop(loop_factory, fast, loop_debug):  # type: ignore[no-untyped-def]
+def loop(
+    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)  # pragma: no cover
         asyncio.set_event_loop(_loop)
         yield _loop
 
 
 @pytest.fixture
-def proactor_loop():  # type: ignore[no-untyped-def]
-    policy = asyncio.WindowsProactorEventLoopPolicy()  # type: ignore[attr-defined]
-    asyncio.set_event_loop_policy(policy)
+def proactor_loop() -> Iterator[asyncio.AbstractEventLoop]:
+    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

@@ -230,21 +230,17 @@ def _proto_factory(conn_closing_result=None, **kwargs):
 
 
 @pytest.fixture
-def selector_loop():
-    policy = asyncio.WindowsSelectorEventLoopPolicy()
-    asyncio.set_event_loop_policy(policy)
-
-    with loop_context(policy.new_event_loop) as _loop:
+def selector_loop() -> Iterator[asyncio.AbstractEventLoop]:
+    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

@@ -107,7 +107,7 @@ def create_mocked_conn(conn_closing_result=None, **kwargs):
     try:
         loop = asyncio.get_running_loop()
     except RuntimeError:
-        loop = asyncio.get_event_loop_policy().get_event_loop()
+        loop = asyncio.get_event_loop()
 
     proto = mock.Mock(**kwargs)
     proto.closed = loop.create_future()

tests/test_loop.py

@@ -37,8 +37,8 @@ def test_default_loop(self) -> None:
         self.assertIs(self.loop, asyncio.get_event_loop_policy().get_event_loop())
 
 
-def test_default_loop(loop) -> None:
-    assert asyncio.get_event_loop_policy().get_event_loop() is loop
+def test_default_loop(loop: asyncio.AbstractEventLoop) -> None:
+    assert asyncio.get_event_loop() is loop
 
 
 def test_setup_loop_non_main_thread() -> None:
@@ -47,7 +47,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

@@ -204,7 +204,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",
@@ -216,6 +215,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()