remove simple test loop

Author: maxfischer2781

unittests/test_functools.py

@@ -5,7 +5,7 @@
 import asyncstdlib as a
 from asyncstdlib.functools import CachedProperty
 
-from .utility import Lock, Schedule, Switch, asyncify, multi_sync, sync
+from .utility import Lock, Schedule, Switch, asyncify, sync
 
 
 @sync
@@ -44,7 +44,7 @@ async def bar(self):
         Foo().bar
 
 
-@multi_sync
+@sync
 async def test_cache_property_order():
     class Value:
         def __init__(self, value):
@@ -66,7 +66,7 @@ async def check_increment(to):
     assert (await val.cached) == 1337  # last value fetched
 
 
-@multi_sync
+@sync
 async def test_cache_property_lock_order():
     class Value:
         def __init__(self, value):
@@ -87,7 +87,7 @@ async def check_cached(to, expected):
     assert (await val.cached) == 5  # first value fetched
 
 
-@multi_sync
+@sync
 async def test_cache_property_lock_deletion():
     class Value:
         def __init__(self, value):
@@ -300,7 +300,7 @@ async def pingpong(arg):
 
 
 @pytest.mark.parametrize("size", [16, None])
-@multi_sync
+@sync
 async def test_lru_cache_concurrent(size):
     current = 0
 

unittests/test_itertools.py

@@ -6,7 +6,7 @@
 
 import asyncstdlib as a
 
-from .utility import sync, asyncify, awaitify, multi_sync, Schedule, Switch, Lock
+from .utility import sync, asyncify, awaitify, Schedule, Switch, Lock
 
 
 @sync
@@ -314,7 +314,7 @@ async def test_tee():
             assert await a.list(iterator) == iterable
 
 
-@multi_sync
+@sync
 async def test_tee_concurrent_locked():
     """Test that properly uses a lock for synchronisation"""
     items = [1, 2, 3, -5, 12, 78, -1, 111]
@@ -344,7 +344,7 @@ async def test_peer(peer_tee):
     platform.python_implementation() != "CPython",
     reason="async generators only protect against concurrent access on CPython",
 )
-@multi_sync
+@sync
 async def test_tee_concurrent_unlocked():
     """Test that tee does not prevent concurrency without a lock"""
     items = list(range(12))

unittests/utility.py

@@ -64,34 +64,6 @@ async def inside_loop() -> bool:
     return await signal is signal
 
 
-def sync(test_case: Callable[..., Coroutine[None, Any, Any]]) -> Callable[..., None]:
-    """
-    Mark an ``async def`` test case to be run synchronously
-
-    This emulates a primitive "event loop" which only responds
-    to the :py:class:`PingPong` by sending it back. This loop
-    is appropriate for tests that do not check concurrency.
-    """
-
-    @wraps(test_case)
-    def run_sync(*args: Any, **kwargs: Any) -> None:
-        coro = test_case(*args, **kwargs)
-        try:
-            event = None
-            while True:
-                event = coro.send(event)
-                if not isinstance(event, PingPong):  # pragma: no cover
-                    raise RuntimeError(
-                        f"test case {test_case} yielded an unexpected event {event}"
-                    )
-        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:
     r"""
     Signal to the event loop to adopt and run new coroutines
@@ -159,9 +131,7 @@ async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any):
         self._owned = False
 
 
-def multi_sync(
-    test_case: Callable[..., Coroutine[None, Any, Any]], /
-) -> Callable[..., None]:
+def sync(test_case: Callable[..., Coroutine[None, Any, Any]], /) -> Callable[..., None]:
     """
     Mark an ``async def`` test case to be run synchronously with children