update docs and types

Author: maxfischer2781

unittests/utility.py

@@ -1,6 +1,7 @@
 from typing import (
     Callable,
     Coroutine,
+    Generator,
     Iterable,
     AsyncIterator,
     TypeVar,
@@ -17,32 +18,52 @@
 
 
 async def asyncify(iterable: Iterable[T]) -> AsyncIterator[T]:
-    """Convert an iterable to async iterable"""
+    """
+    Convert an iterable into an async iterable
+
+    This is intended to sequence literals like lists to `async` iterators
+    in order to force usage of `async` code paths. There is no functional
+    or other advantage otherwise.
+    """
     for value in iterable:
         yield value
 
 
 def awaitify(call: Callable[..., T]) -> Callable[..., Awaitable[T]]:
-    async def await_wrapper(*args, **kwargs):
+    """
+    Convert a callable (`foo()`) into an async callable (`await foo()`)
+
+    This is intended to convert `lambda` expressions to `async` functions
+    in order to force usage of `async` code paths. There is no functional
+    or other advantage otherwise.
+    """
+
+    async def await_wrapper(*args: Any, **kwargs: Any) -> T:
         return call(*args, **kwargs)
 
     return await_wrapper
 
 
 class PingPong:
-    """Signal to the event loop which gets returned unchanged"""
+    """
+    Signal to the event loop which gets returned unchanged
 
-    def __await__(self):
+    The coroutine yields to the event loop but is resumed
+    immediately, without running others in the meantime.
+    This is mainly useful for debugging the event loop.
+    """
+
+    def __await__(self) -> "Generator[PingPong, Any, Any]":
         return (yield self)
 
 
-async def inside_loop():
+async def inside_loop() -> bool:
     """Test whether there is an active event loop available"""
     signal = PingPong()
     return await signal is signal
 
 
-def sync(test_case: Callable[..., Coroutine[T, Any, Any]]) -> Callable[..., T]:
+def sync(test_case: Callable[..., Coroutine[None, Any, Any]]) -> Callable[..., None]:
     """
     Mark an ``async def`` test case to be run synchronously
 
@@ -51,7 +72,7 @@ def sync(test_case: Callable[..., Coroutine[T, Any, Any]]) -> Callable[..., T]:
     """
 
     @wraps(test_case)
-    def run_sync(*args: Any, **kwargs: Any) -> T:
+    def run_sync(*args: Any, **kwargs: Any) -> None:
         coro = test_case(*args, **kwargs)
         try:
             event = None
@@ -63,13 +84,21 @@ def run_sync(*args: Any, **kwargs: Any) -> T:
                     )
         except StopIteration as e:
             result = e.args[0] if e.args else None
+            assert result is None, f"got '{result!r}' expected 'None'"
         return result
 
     return run_sync
 
 
 class Schedule:
-    """Signal to the event loop to adopt and run a new coroutine"""
+    r"""
+    Signal to the event loop to adopt and run new coroutines
+
+    :param coros: The coroutines to start running
+
+    In order to communicate with the event loop and start the coroutines,
+    the :py:class:`Schedule` must be `await`\ ed.
+    """
 
     def __init__(self, *coros: Coroutine[Any, Any, Any]):
         self.coros = coros
@@ -79,13 +108,22 @@ def __await__(self):
 
 
 class Switch:
-    """Signal to the event loop to run another coroutine"""
+    """
+    Signal to the event loop to run another coroutine
+
+    Pauses the coroutine but immediately continues after
+    all other runnable coroutines of the event loop.
+    This is similar to the common ``sleep(0)`` function
+    of regular event loop frameworks.
+    """
 
     def __await__(self):
         yield self
 
 
 class Lock:
+    """Simple lock for exclusive access"""
+
     def __init__(self):
         self._owned = False
         self._waiting: list[object] = []
@@ -95,18 +133,20 @@ async def __aenter__(self):
             # wait until it is our turn to take the lock
             token = object()
             self._waiting.append(token)
+            # a spin-lock should be fine since tests are short anyways
             while self._owned or self._waiting[0] is not token:
                 await Switch()
-            # take the lock and remove our wait claim
-            self._owned = True
+            # we will take the lock now, remove our wait claim
             self._waiting.pop(0)
         self._owned = True
 
     async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any):
         self._owned = False
 
 
-def multi_sync(test_case: Callable[..., Coroutine[T, Any, Any]]) -> Callable[..., T]:
+def multi_sync(
+    test_case: Callable[..., Coroutine[None, Any, Any]]
+) -> Callable[..., None]:
     """
     Mark an ``async def`` test case to be run synchronously with children