Create asyncioEventLoop.runCoroutineSync helper function to call a coroutine, wait for the result and either return that result or any raised exceptions

Author: bramd

source/asyncioEventLoop.py

@@ -58,6 +58,30 @@ def runCoroutine(coro: Coroutine) -> asyncio.Future:
 
 	:param coro: The coroutine to run.
 	"""
-	if not asyncioThread.is_alive():
+	if asyncioThread is None or not asyncioThread.is_alive():
 		raise RuntimeError("Asyncio event loop thread is not running")
 	return asyncio.run_coroutine_threadsafe(coro, eventLoop)
+
+
+def runCoroutineSync(coro: Coroutine, timeout: float | None = None):
+	"""Schedule a coroutine to be run on the asyncio event loop and wait for the result.
+
+	This is a synchronous wrapper around runCoroutine() that blocks until the coroutine
+	completes and returns the result directly, or raises any exception that occurred.
+
+	:param coro: The coroutine to run.
+	:param timeout: Optional timeout in seconds. If None, waits indefinitely.
+	:return: The result of the coroutine.
+	:raises: Any exception raised by the coroutine.
+	:raises TimeoutError: If the timeout is exceeded.
+	:raises RuntimeError: If the asyncio event loop thread is not running.
+	"""
+	future = runCoroutine(coro)
+	try:
+		# Wait for the future to complete and get the result
+		# This will raise any exception that occurred in the coroutine
+		return future.result(timeout)
+	except asyncio.TimeoutError as e:
+		# Cancel the coroutine since it timed out
+		future.cancel()
+		raise TimeoutError(f"Coroutine execution timed out after {timeout} seconds") from e

source/hwIo/ble/_io.py

@@ -9,7 +9,7 @@
 from typing import Callable, Iterator
 import weakref
 
-from asyncioEventLoop import runCoroutine
+from asyncioEventLoop import runCoroutineSync
 from ..base import _isDebug, IoBase
 from ..ioThread import IoThread
 from logHandler import log
@@ -120,10 +120,7 @@ def __init__(
 			daemon=True,
 		)
 		self._readerThread.start()
-		f = runCoroutine(self._initAndConnect())
-		f.result()
-		if f.exception():
-			raise f.exception()
+		runCoroutineSync(self._initAndConnect(), timeout=CONNECT_TIMEOUT_SECONDS)
 		self.waitForConnection(CONNECT_TIMEOUT_SECONDS)
 
 	async def _initAndConnect(self) -> None:
@@ -157,19 +154,16 @@ def write(self, data: bytes):
 
 		# Split the data into chunks that fit within the MTU
 		for s in sliced(data, characteristic.max_write_without_response_size):
-			f = runCoroutine(
+			runCoroutineSync(
 				self._client.write_gatt_char(characteristic, s, response=False),
 			)
-			f.result()
-			if f.exception():
-				raise f.exception()
 
 	def close(self) -> None:
 		"""Disconnect the BLE peripheral and release resources."""
 		if _isDebug():
 			log.debug("Closing BLE connection")
 		if self._client.is_connected:
-			runCoroutine(self._client.disconnect()).result()
+			runCoroutineSync(self._client.disconnect())
 		self._queuedData.join()
 		self._stopReaderEvent.set()
 		self._readerThread.join()

tests/unit/test_asyncioEventLoop.py

@@ -0,0 +1,95 @@
+# A part of NonVisual Desktop Access (NVDA)
+# Copyright (C) 2025 NV Access Limited, Bram Duvigneau, Dot Incorporated
+# This file is covered by the GNU General Public License.
+# See the file COPYING for more details.
+
+"""Unit tests for asyncioEventLoop module."""
+
+import asyncio
+import unittest
+
+import asyncioEventLoop
+
+
+class TestRunCoroutineSync(unittest.TestCase):
+	"""Tests for runCoroutineSync function."""
+
+	@classmethod
+	def setUpClass(cls):
+		"""Initialize the asyncio event loop before tests."""
+		asyncioEventLoop.initialize()
+
+	@classmethod
+	def tearDownClass(cls):
+		"""Terminate the asyncio event loop after tests."""
+		asyncioEventLoop.terminate()
+
+	def test_returnsResult(self):
+		"""Test that runCoroutineSync returns the coroutine's result."""
+
+		async def simpleCoroutine():
+			return 42
+
+		result = asyncioEventLoop.runCoroutineSync(simpleCoroutine())
+		self.assertEqual(result, 42)
+
+	def test_returnsComplexResult(self):
+		"""Test that runCoroutineSync returns complex objects."""
+
+		async def complexCoroutine():
+			await asyncio.sleep(0.01)
+			return {"key": "value", "number": 123}
+
+		result = asyncioEventLoop.runCoroutineSync(complexCoroutine())
+		self.assertEqual(result, {"key": "value", "number": 123})
+
+	def test_raisesException(self):
+		"""Test that runCoroutineSync raises exceptions from the coroutine."""
+
+		async def failingCoroutine():
+			await asyncio.sleep(0.01)
+			raise ValueError("Test error message")
+
+		with self.assertRaises(ValueError) as cm:
+			asyncioEventLoop.runCoroutineSync(failingCoroutine())
+		self.assertEqual(str(cm.exception), "Test error message")
+
+	def test_timeoutRaisesTimeoutError(self):
+		"""Test that runCoroutineSync raises TimeoutError when timeout is exceeded."""
+
+		async def slowCoroutine():
+			await asyncio.sleep(10)
+			return "Should not reach here"
+
+		with self.assertRaises(TimeoutError) as cm:
+			asyncioEventLoop.runCoroutineSync(slowCoroutine(), timeout=0.1)
+		self.assertIn("timed out", str(cm.exception).lower())
+
+	def test_noTimeoutWaitsIndefinitely(self):
+		"""Test that runCoroutineSync waits indefinitely when no timeout is specified."""
+
+		async def delayedCoroutine():
+			await asyncio.sleep(0.1)
+			return "completed"
+
+		# This should complete successfully even though it takes some time
+		result = asyncioEventLoop.runCoroutineSync(delayedCoroutine())
+		self.assertEqual(result, "completed")
+
+	def test_raisesRuntimeErrorWhenEventLoopNotRunning(self):
+		"""Test that runCoroutineSync raises RuntimeError when event loop is not running."""
+		# Save original thread reference
+		originalThread = asyncioEventLoop.asyncioThread
+
+		# Temporarily set to None to simulate not running
+		asyncioEventLoop.asyncioThread = None
+
+		async def anyCoroutine():
+			return "test"
+
+		with self.assertRaises(RuntimeError) as cm:
+			asyncioEventLoop.runCoroutineSync(anyCoroutine())
+		self.assertIn("not running", str(cm.exception).lower())
+
+		# Restore original thread
+		asyncioEventLoop.asyncioThread = originalThread

tests/unit/test_hwIo_ble.py

@@ -216,13 +216,10 @@ def setUp(self):
 		self.mockServices.services = mockServicesDict
 		self.mockClient.services = self.mockServices
 
-		# Patch runCoroutine
-		self.runCoroutinePatcher = patch("hwIo.ble._io.runCoroutine")
-		self.mockRunCoroutine = self.runCoroutinePatcher.start()
-		mockFuture = MagicMock()
-		mockFuture.result.return_value = None
-		mockFuture.exception.return_value = None
-		self.mockRunCoroutine.return_value = mockFuture
+		# Patch runCoroutineSync (just returns None, as it's a synchronous wrapper)
+		self.runCoroutineSyncPatcher = patch("hwIo.ble._io.runCoroutineSync")
+		self.mockRunCoroutineSync = self.runCoroutineSyncPatcher.start()
+		self.mockRunCoroutineSync.return_value = None
 
 		# Import Ble after patching
 		from hwIo.ble._io import Ble
@@ -231,7 +228,7 @@ def setUp(self):
 
 	def tearDown(self):
 		"""Clean up patches."""
-		self.runCoroutinePatcher.stop()
+		self.runCoroutineSyncPatcher.stop()
 		self.bleakClientPatcher.stop()
 
 	def test_connectionSuccess(self):
@@ -265,8 +262,8 @@ def onReceive(data):
 		callArgs = self.mockBleakClientClass.call_args
 		self.assertEqual(callArgs[0][0], mockDevice)
 
-		# Verify runCoroutine was called at least once (_initAndConnect contains connect + start_notify)
-		self.assertGreaterEqual(self.mockRunCoroutine.call_count, 1)
+		# Verify runCoroutineSync was called for initialization
+		self.mockRunCoroutineSync.assert_called()
 
 		# Verify the connection is established
 		self.assertTrue(ble.isConnected())
@@ -296,8 +293,8 @@ def test_writeData(self):
 		self.mockServices.get_service.assert_called_with("service-uuid")
 		self.mockService.get_characteristic.assert_called_with("write-char-uuid")
 
-		# Verify write was called (through runCoroutine)
-		self.mockRunCoroutine.assert_called()
+		# Verify write was called (through runCoroutineSync)
+		self.assertGreater(self.mockRunCoroutineSync.call_count, 1)  # At least init + write
 
 	def test_writeDataChunking(self):
 		"""Test that large data is split into MTU-sized chunks."""
@@ -320,14 +317,14 @@ def test_writeDataChunking(self):
 		)
 
 		# Reset call count after initialization
-		initialCallCount = self.mockRunCoroutine.call_count
+		initialCallCount = self.mockRunCoroutineSync.call_count
 
 		# Write 25 bytes (should split into 3 chunks: 10, 10, 5)
 		testData = b"A" * 25
 		ble.write(testData)
 
-		# Verify runCoroutine was called 3 times for writes (plus initial calls)
-		writeCalls = self.mockRunCoroutine.call_count - initialCallCount
+		# Verify runCoroutineSync was called 3 times for writes
+		writeCalls = self.mockRunCoroutineSync.call_count - initialCallCount
 		self.assertEqual(writeCalls, 3)
 
 	def test_receiveNotification(self):
@@ -353,9 +350,8 @@ def onReceive(data):
 			ioThread=mockIoThread,
 		)
 
-		# Get the notification callback that was registered
-		# It should be in the call to start_notify via runCoroutine
-		self.assertTrue(self.mockRunCoroutine.called)
+		# Verify initialization occurred
+		self.mockRunCoroutineSync.assert_called()
 
 		# Simulate notification by calling _notifyReceive directly
 		testData = bytearray(b"notification data")
@@ -385,8 +381,8 @@ def test_closeCleanup(self):
 		# Close the connection
 		ble.close()
 
-		# Verify disconnect was called via runCoroutine
-		self.assertTrue(self.mockRunCoroutine.called)
+		# Verify disconnect was called via runCoroutineSync (init + close)
+		self.assertGreater(self.mockRunCoroutineSync.call_count, 1)
 
 		# Verify onReceive callback was cleared
 		self.assertIsNone(ble._onReceive)