Use asyncio loop_factory instead of policies for python>=3.12

qasync/__init__.py

@@ -849,28 +849,33 @@ def wrapper(*args, **kwargs):
     return outer_decorator
 
 
-class QEventLoopPolicyMixin:
-    def new_event_loop(self):
-        return QEventLoop(QApplication.instance() or QApplication(sys.argv))
+def _get_qevent_loop():
+    return QEventLoop(QApplication.instance() or QApplication(sys.argv))
 
 
-class DefaultQEventLoopPolicy(
-    QEventLoopPolicyMixin,
-    asyncio.DefaultEventLoopPolicy,
-):
-    pass
-
-
-@contextlib.contextmanager
-def _set_event_loop_policy(policy):
-    old_policy = asyncio.get_event_loop_policy()
-    asyncio.set_event_loop_policy(policy)
-    try:
-        yield
-    finally:
-        asyncio.set_event_loop_policy(old_policy)
+if sys.version_info >= (3, 12):
 
+    def run(*args, **kwargs):
+        return asyncio.run(
+            *args,
+            **kwargs,
+            loop_factory=_get_qevent_loop,
+        )
+else:
+    # backwards compatibility with event loop policies
+    class DefaultQEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+        def new_event_loop(self):
+            return _get_qevent_loop()
+
+    @contextlib.contextmanager
+    def _set_event_loop_policy(policy):
+        old_policy = asyncio.get_event_loop_policy()
+        asyncio.set_event_loop_policy(policy)
+        try:
+            yield
+        finally:
+            asyncio.set_event_loop_policy(old_policy)
 
-def run(*args, **kwargs):
-    with _set_event_loop_policy(DefaultQEventLoopPolicy()):
-        return asyncio.run(*args, **kwargs)
+    def run(*args, **kwargs):
+        with _set_event_loop_policy(DefaultQEventLoopPolicy()):
+            return asyncio.run(*args, **kwargs)

tests/test_qeventloop.py

@@ -105,7 +105,7 @@ def test_can_handle_exception_in_executor(self, loop, executor):
             loop.run_until_complete(
                 asyncio.wait_for(
                     loop.run_in_executor(executor, self.blocking_failure),
-                    timeout=3.0,
+                    timeout=10.0,
                 )
             )
 
@@ -126,7 +126,7 @@ def blocking_func(self, was_invoked):
     async def blocking_task(self, loop, executor, was_invoked):
         logging.debug("start blocking task()")
         fut = loop.run_in_executor(executor, self.blocking_func, was_invoked)
-        await asyncio.wait_for(fut, timeout=5.0)
+        await asyncio.wait_for(fut, timeout=10.0)
         logging.debug("start blocking task()")
 
 
@@ -140,7 +140,7 @@ async def mycoro():
         await process.wait()
         assert process.returncode == 5
 
-    loop.run_until_complete(asyncio.wait_for(mycoro(), timeout=3))
+    loop.run_until_complete(asyncio.wait_for(mycoro(), timeout=10.0))
 
 
 def test_can_read_subprocess(loop):
@@ -160,7 +160,7 @@ async def mycoro():
         assert process.returncode == 0
         assert received_stdout.strip() == b"Hello async world!"
 
-    loop.run_until_complete(asyncio.wait_for(mycoro(), timeout=3))
+    loop.run_until_complete(asyncio.wait_for(mycoro(), timeout=10.0))
 
 
 def test_can_communicate_subprocess(loop):
@@ -181,7 +181,7 @@ async def mycoro():
         assert process.returncode == 0
         assert received_stdout.strip() == b"Hello async world!"
 
-    loop.run_until_complete(asyncio.wait_for(mycoro(), timeout=3))
+    loop.run_until_complete(asyncio.wait_for(mycoro(), timeout=10.0))
 
 
 def test_can_terminate_subprocess(loop):
@@ -809,103 +809,6 @@ async def mycoro():
     assert "seconds" in msg
 
 
-@pytest.mark.skipif(sys.version_info < (3, 12), reason="Requires Python 3.12+")
-def test_asyncio_run(application):
-    """Test that QEventLoop is compatible with asyncio.run()"""
-    done = False
-    loop = None
-
-    async def main():
-        nonlocal done, loop
-        assert loop.is_running()
-        assert asyncio.get_running_loop() is loop
-        await asyncio.sleep(0.01)
-        done = True
-
-    def factory():
-        nonlocal loop
-        loop = qasync.QEventLoop(application)
-        return loop
-
-    asyncio.run(main(), loop_factory=factory)
-    assert done
-    assert loop.is_closed()
-    assert not loop.is_running()
-
-
-@pytest.mark.skipif(sys.version_info < (3, 12), reason="Requires Python 3.12+")
-def test_asyncio_run_cleanup(application):
-    """Test that running tasks are cleaned up"""
-    task = None
-    cancelled = False
-
-    async def main():
-        nonlocal task, cancelled
-
-        async def long_task():
-            nonlocal cancelled
-            try:
-                await asyncio.sleep(10)
-            except asyncio.CancelledError:
-                cancelled = True
-
-        task = asyncio.create_task(long_task())
-        await asyncio.sleep(0.01)
-
-    asyncio.run(main(), loop_factory=lambda: qasync.QEventLoop(application))
-    assert cancelled
-
-
-def test_qasync_run(application):
-    """Test running with qasync.run()"""
-    done = False
-    loop = None
-
-    async def main():
-        nonlocal done, loop
-        loop = asyncio.get_running_loop()
-        assert loop.is_running()
-        await asyncio.sleep(0.01)
-        done = True
-
-    # qasync.run uses an EventLoopPolicy to create the loop
-    qasync.run(main())
-    assert done
-    assert loop.is_closed()
-    assert not loop.is_running()
-
-
-def test_qeventloop_in_qthread():
-    class CoroutineExecutorThread(qasync.QtCore.QThread):
-        def __init__(self, coro):
-            super().__init__()
-            self.coro = coro
-            self.loop = None
-
-        def run(self):
-            self.loop = qasync.QEventLoop(self)
-            asyncio.set_event_loop(self.loop)
-            asyncio.run(self.coro)
-
-        def join(self):
-            self.loop.stop()
-            self.loop.close()
-            self.wait()
-
-    event = threading.Event()
-
-    async def coro():
-        await asyncio.sleep(0.1)
-        event.set()
-
-    thread = CoroutineExecutorThread(coro())
-    thread.start()
-
-    assert event.wait(timeout=1), "Coroutine did not execute successfully"
-
-    thread.join()  # Ensure thread cleanup
-
-
 def teardown_module(module):
     """
     Remove handlers from all loggers

tests/test_run.py

@@ -1,4 +1,5 @@
 import asyncio
+import sys
 from unittest.mock import ANY
 
 import pytest
@@ -25,6 +26,7 @@ def test_qasync_run_restores_loop(get_event_loop_coro):
         _ = asyncio.get_event_loop()
 
 
+@pytest.mark.skipif(sys.version_info >= (3, 14), reason="Deprecated since Python 3.14")
 def test_qasync_run_restores_policy(get_event_loop_coro):
     old_policy = asyncio.get_event_loop_policy()
     qasync.run(get_event_loop_coro(ANY))
@@ -35,3 +37,69 @@ def test_qasync_run_restores_policy(get_event_loop_coro):
 def test_qasync_run_with_debug_args(get_event_loop_coro):
     qasync.run(get_event_loop_coro(True), debug=True)
     qasync.run(get_event_loop_coro(False), debug=False)
+
+
+@pytest.mark.skipif(sys.version_info < (3, 12), reason="Requires Python 3.12+")
+def test_asyncio_run(application):
+    """Test that QEventLoop is compatible with asyncio.run()"""
+    done = False
+    loop = None
+
+    async def main():
+        nonlocal done, loop
+        assert loop.is_running()
+        assert asyncio.get_running_loop() is loop
+        await asyncio.sleep(0.01)
+        done = True
+
+    def factory():
+        nonlocal loop
+        loop = qasync.QEventLoop(application)
+        return loop
+
+    asyncio.run(main(), loop_factory=factory)
+    assert done
+    assert loop.is_closed()
+    assert not loop.is_running()
+
+
+@pytest.mark.skipif(sys.version_info < (3, 12), reason="Requires Python 3.12+")
+def test_asyncio_run_cleanup(application):
+    """Test that running tasks are cleaned up"""
+    task = None
+    cancelled = False
+
+    async def main():
+        nonlocal task, cancelled
+
+        async def long_task():
+            nonlocal cancelled
+            try:
+                await asyncio.sleep(10)
+            except asyncio.CancelledError:
+                cancelled = True
+
+        task = asyncio.create_task(long_task())
+        await asyncio.sleep(0.01)
+
+    asyncio.run(main(), loop_factory=lambda: qasync.QEventLoop(application))
+    assert cancelled
+
+
+def test_qasync_run(application):
+    """Test running with qasync.run()"""
+    done = False
+    loop = None
+
+    async def main():
+        nonlocal done, loop
+        loop = asyncio.get_running_loop()
+        assert loop.is_running()
+        await asyncio.sleep(0.01)
+        done = True
+
+    # qasync.run uses an EventLoopPolicy to create the loop
+    qasync.run(main())
+    assert done
+    assert loop.is_closed()
+    assert not loop.is_running()