Loading tests/_anyio.py +24 −3 Original line number Diff line number Diff line Loading @@ -21,21 +21,35 @@ from functools import wraps from typing import Any from typing import Callable from typing import Coroutine from typing import Literal from typing import Tuple from typing import Union from unittest import TestCase from unittest import mock from warnings import warn import anyio try: import trio as _ # noqa USE_TRIO = True except ImportError: USE_TRIO = False Backend = Union[Literal['asyncio'], Literal['trio']] py_version = sys.version_info[:2] AsyncTestFunc = Callable[..., Coroutine[Any, Any, None]] TestFunc = Callable[..., None] def with_anyio(timeout: int = 10) -> Callable[[AsyncTestFunc], TestFunc]: def with_anyio(*backends: Backend, timeout: int = 10) -> Callable[[AsyncTestFunc], TestFunc]: """ Create a wrapping decorator to run asynchronous test functions """ if not backends: backends = ('asyncio',) def decorator(testfunc: AsyncTestFunc) -> TestFunc: async def test_async_wrapper(tc: TestCase, args: Tuple[mock.Mock]) -> None: Loading @@ -43,8 +57,15 @@ def with_anyio(timeout: int = 10) -> Callable[[AsyncTestFunc], TestFunc]: await testfunc(tc, *args) @wraps(testfunc) def test_wrapper(*args: mock.Mock) -> None: anyio.run(test_async_wrapper, args) def test_wrapper(tc: TestCase, *args: mock.Mock) -> None: for backend in backends: if backend == 'trio' and not USE_TRIO: warn( f"not running {testfunc.__name__} with trio; package is missing", ) continue with tc.subTest(f"backend: {backend}"): anyio.run(test_async_wrapper, tc, args, backend=backend) return test_wrapper Loading Loading
tests/_anyio.py +24 −3 Original line number Diff line number Diff line Loading @@ -21,21 +21,35 @@ from functools import wraps from typing import Any from typing import Callable from typing import Coroutine from typing import Literal from typing import Tuple from typing import Union from unittest import TestCase from unittest import mock from warnings import warn import anyio try: import trio as _ # noqa USE_TRIO = True except ImportError: USE_TRIO = False Backend = Union[Literal['asyncio'], Literal['trio']] py_version = sys.version_info[:2] AsyncTestFunc = Callable[..., Coroutine[Any, Any, None]] TestFunc = Callable[..., None] def with_anyio(timeout: int = 10) -> Callable[[AsyncTestFunc], TestFunc]: def with_anyio(*backends: Backend, timeout: int = 10) -> Callable[[AsyncTestFunc], TestFunc]: """ Create a wrapping decorator to run asynchronous test functions """ if not backends: backends = ('asyncio',) def decorator(testfunc: AsyncTestFunc) -> TestFunc: async def test_async_wrapper(tc: TestCase, args: Tuple[mock.Mock]) -> None: Loading @@ -43,8 +57,15 @@ def with_anyio(timeout: int = 10) -> Callable[[AsyncTestFunc], TestFunc]: await testfunc(tc, *args) @wraps(testfunc) def test_wrapper(*args: mock.Mock) -> None: anyio.run(test_async_wrapper, args) def test_wrapper(tc: TestCase, *args: mock.Mock) -> None: for backend in backends: if backend == 'trio' and not USE_TRIO: warn( f"not running {testfunc.__name__} with trio; package is missing", ) continue with tc.subTest(f"backend: {backend}"): anyio.run(test_async_wrapper, tc, args, backend=backend) return test_wrapper Loading