Test run_forever and run_until_complete return values

Author: hosaka

tests/conftest.py

@@ -1,12 +1,15 @@
-# © 2018 Gerard Marull-Paretas <gerard@teslabs.com>
-# © 2014 Mark Harviston <mark.harviston@gmail.com>
-# © 2014 Arve Knudsen <arve.knudsen@gmail.com>
-# BSD License
+"""
+Copyright (c) 2018 Gerard Marull-Paretas <gerard@teslabs.com>
+Copyright (c) 2014 Mark Harviston <mark.harviston@gmail.com>
+Copyright (c) 2014 Arve Knudsen <arve.knudsen@gmail.com>
+
+BSD License
+"""
 
-import os
 import logging
-from pytest import fixture
+import os
 
+from pytest import fixture
 
 logging.basicConfig(
     level=logging.DEBUG, format="%(levelname)s\t%(filename)s:%(lineno)s %(message)s"

tests/test_qeventloop.py

@@ -5,13 +5,13 @@
 
 import asyncio
 import ctypes
-import threading
 import logging
 import multiprocessing
 import os
 import socket
 import subprocess
 import sys
+import threading
 import time
 from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
 
@@ -809,6 +809,23 @@ async def mycoro():
     assert "seconds" in msg
 
 
+def test_run_until_complete_returns_future_result(loop):
+    async def coro():
+        await asyncio.sleep(0)
+        return 42
+
+    assert loop.run_until_complete(asyncio.wait_for(coro(), timeout=0.1)) == 42
+
+
+def test_run_forever_custom_exit_code(loop, application):
+    orig_exec = application.exec
+    application.exec = lambda: 42
+    try:
+        assert loop.run_forever() == 42
+    finally:
+        application.exec = orig_exec
+
+
 @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()"""