Improve coroutine timeout handling (#295)

resolves #294

This PR...

1. adds class-level `timeout_ms` attribute to customize runtime of asyncio coroutines.

   Example:

   ```py
      class MyAsyncTestCase(AsyncTestCase):
          timeout_ms=10000
   ```

2. uses `timeout_ms` argument from test_... coroutines to cutomize test case runtime.

   Example:

   ```py
      async def test_anything(self, timeout_ms=10000):
          ...
   ```

3. raises TimeoutError within context of currently awaited asynchronous coroutine to ensure valuable tracebacks being printed.

   Example:

   ```
   test_coroutine (test_asyncio_timeout.MyAsyncTestCaseA.test_coroutine) ... ERROR
   
   ======================================================================
   ERROR: test_coroutine (test_asyncio_timeout.MyAsyncTestCaseA.test_coroutine)
   ----------------------------------------------------------------------
   Traceback (most recent call last):
     File "<ST>\Data\Packages\UnitTesting\unittesting\core\py313\case.py", line 120, in run
       outcome.expecting_failure = False
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
     File "<ST>\Data\Packages\UnitTesting\unittesting\core\py313\case.py", line 71, in _callMaybeCoro
     File "<ST>\Data\Packages\UnitTesting\tests\test_asyncio_timeout.py", line 23, in test_coroutine
       await a_coro(self)
     File "<ST>\Data\Packages\UnitTesting\tests\test_asyncio_timeout.py", line 7, in a_coro
       await a_coro2(test)
     File "<ST>\Data\Packages\UnitTesting\tests\test_asyncio_timeout.py", line 11, in a_coro2
       await a_coro3(test)
     File "<ST>\Data\Packages\UnitTesting\tests\test_asyncio_timeout.py", line 15, in a_coro3
       await asyncio.sleep(10.0)
     File "./python3.14/asyncio/tasks.py", line 702, in sleep
   TimeoutError: Task not completed within 4.50 seconds.
   
   ----------------------------------------------------------------------
   Ran 1 test in 4.667s
   
   FAILED (errors=1)
   ```

Author: deathaxe

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/unittesting.json

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

tests/_Asyncio/.python-version tests/_Asyncio_Success/.python-versiontests/_Asyncio/.python-version renamed to tests/_Asyncio_Success/.python-version

@@ -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/_Asyncio/tests/test_coroutine.py …_Asyncio_Success/tests/test_coroutine.pytests/_Asyncio/tests/test_coroutine.py renamed to tests/_Asyncio_Success/tests/test_coroutine.py tests/_Asyncio/tests/test_coroutine.py renamed to tests/_Asyncio_Success/tests/test_coroutine.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 = (

tests/_Asyncio_Timeout/.python-version

@@ -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):

tests/_Asyncio_Timeout/tests/test_coroutine.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):