Improve coroutine timeout handling

README.md

@@ -590,6 +590,8 @@ async def async_coroutine(view):
 
 
 class MyAsyncTestCase(AsyncTestCase):
+    timeout_ms = 4000
+    """Class wide coroutine timeout."""
 
     @classmethod
     async def setUpClass(cls):
@@ -613,7 +615,10 @@ class MyAsyncTestCase(AsyncTestCase):
           "Initial Content"
         )
 
-    async def test_coroutine(self):
+    async def test_coroutine(self, timeout_ms=10000):
+        """
+        A long running coroutine with custom timeout.
+        """
         await async_coroutine(self.view)
         self.assertEqual(
           self.view.substr(sublime.Region(0, self.view.size())),

tests/_Asyncio_Timeout/.python-version

@@ -0,0 +1 @@
+3.8

tests/_Asyncio_Timeout/tests/test_coroutine.py

@@ -0,0 +1,19 @@
+import asyncio
+
+from unittesting import AsyncTestCase
+
+
+async def a_coro(test):
+    await asyncio.sleep(1.0)
+
+
+class MyAsyncTestCaseA(AsyncTestCase):
+    timeout_ms = 100
+
+    async def test_coroutine_class_timeout(self):
+        await a_coro(self)
+
+
+class MyAsyncTestCaseB(AsyncTestCase):
+    async def test_coroutine_local_timeout(self, timeout_ms=100):
+        await a_coro(self)

tests/test_3141596.py

@@ -122,7 +122,8 @@ def assertOk(self, txt, msg=None):
 
 class TestUnitTesting(UnitTestingTestCase):
     fixtures = (
-        "_Success", "_Failure", "_Empty", "_Output", "_Deferred", "_Async", "_Asyncio"
+        "_Success", "_Failure", "_Empty", "_Output", "_Deferred", "_Async",
+        "_Asyncio_Success", "_Asyncio_Timeout"
     )
 
     @with_package("_Success")
@@ -154,10 +155,15 @@ def test_async(self, txt):
         self.assertOk(txt)
 
     @skipIf(PY33, "not applicable in Python 3.3")
-    @with_package("_Asyncio")
-    def test_asyncio(self, txt):
+    @with_package("_Asyncio_Success")
+    def test_asyncio_success(self, txt):
         self.assertOk(txt)
 
+    @skipIf(PY33, "not applicable in Python 3.3")
+    @with_package("_Asyncio_Timeout")
+    def test_asyncio_timeout(self, txt):
+        self.assertRegexContains(txt, r'^ERROR')
+
 
 class TestSyntax(UnitTestingTestCase):
     fixtures = (

unittesting/core/py313/case.py

@@ -10,6 +10,7 @@
 from unittest.case import _Outcome
 from unittest.case import expectedFailure
 
+from .runner import DEFAULT_CONDITION_TIMEOUT
 from .runner import defer
 
 __all__ = [
@@ -22,6 +23,7 @@
 
 
 class DeferrableTestCase(TestCase):
+    timeout_ms: int = DEFAULT_CONDITION_TIMEOUT
 
     def _callSetUp(self):
         return self._callMaybeCoro(self.setUp)
@@ -48,15 +50,24 @@ def _callMaybeCoro(cls, func, /, *args, **kwargs):
         elif inspect.iscoroutine(coro):
             fut = cls.run_coroutine(coro)
 
-            def wait_until_complete():
+            def await_future():
                 if not fut.done() and not fut.cancelled():
                     return False
                 exception = fut.exception()
                 if exception is not None:
                     raise exception from None
                 return True
 
-            yield wait_until_complete
+            if frame := coro.cr_frame:
+                # prefer optional timeout from test_... coroutine's arguments
+                timeout_ms = frame.f_locals.get("timeout_ms", cls.timeout_ms)
+            else:
+                timeout_ms = cls.timeout_ms
+            try:
+                yield {"condition": await_future, "timeout": timeout_ms}
+            except TimeoutError:
+                msg = f"Task not completed within {timeout_ms / 1000:.2f} seconds."
+                coro.throw(TimeoutError, msg)
 
     @staticmethod
     def run_coroutine(coro):

unittesting/core/py38/case.py

@@ -9,6 +9,7 @@
 from unittest.case import _Outcome
 from unittest.case import expectedFailure
 
+from .runner import DEFAULT_CONDITION_TIMEOUT
 from .runner import defer
 
 __all__ = [
@@ -21,6 +22,7 @@
 
 
 class DeferrableTestCase(TestCase):
+    timeout_ms: int = DEFAULT_CONDITION_TIMEOUT
 
     def _callSetUp(self):
         return self._callMaybeCoro(self.setUp)
@@ -47,15 +49,24 @@ def _callMaybeCoro(cls, func, /, *args, **kwargs):
         elif inspect.iscoroutine(coro):
             fut = cls.run_coroutine(coro)
 
-            def wait_until_complete():
+            def await_future():
                 if not fut.done() and not fut.cancelled():
                     return False
                 exception = fut.exception()
                 if exception is not None:
                     raise exception from None
                 return True
 
-            yield wait_until_complete
+            if frame := coro.cr_frame:
+                # prefer optional timeout from test_... coroutine's arguments
+                timeout_ms = frame.f_locals.get("timeout_ms", cls.timeout_ms)
+            else:
+                timeout_ms = cls.timeout_ms
+            try:
+                yield {"condition": await_future, "timeout": timeout_ms}
+            except TimeoutError:
+                msg = f"Task not completed within {timeout_ms / 1000:.2f} seconds."
+                coro.throw(TimeoutError, msg)
 
     @staticmethod
     def run_coroutine(coro):