[feat] Added new function "is_async_test" which returns whether a pytest Item is an async test managed by pytest-asyncio.

Signed-off-by: Michael Seifert <[email protected]>

Author: seifertm

docs/source/how-to-guides/index.rst

@@ -7,5 +7,6 @@ How-To Guides
 
   multiple_loops
   uvloop
+  test_item_is_async
 
 This section of the documentation provides code snippets and recipes to accomplish specific tasks with pytest-asyncio.

docs/source/how-to-guides/test_item_is_async.rst

@@ -0,0 +1,7 @@
+=======================================
+How to tell if a test function is async
+=======================================
+Use ``pytest_asyncio.is_async_item`` to determine if a test item is asynchronous and managed by pytest-asyncio.
+
+.. include:: test_item_is_async_example.py
+    :code: python

docs/source/how-to-guides/test_item_is_async_example.py

@@ -0,0 +1,7 @@
+from pytest_asyncio import is_async_test
+
+
+def pytest_collection_modifyitems(items):
+    for item in items:
+        if is_async_test(item):
+            pass

docs/source/reference/changelog.rst

@@ -9,6 +9,7 @@ Changes are non-breaking, unless you upgrade from v0.22.
 
 - BREAKING: The *asyncio_event_loop* mark has been removed. Event loops with class, module, package, and session scopes can be requested via the *scope* keyword argument to the _asyncio_ mark.
 - Introduces the *event_loop_policy* fixture which allows testing with non-default or multiple event loops  `#662 <https://github.com/pytest-dev/pytest-asyncio/pull/662>`_
+- Introduces ``pytest_asyncio.is_async_test`` which returns whether a test item is managed by pytest-asyncio `#376 <https://github.com/pytest-dev/pytest-asyncio/issues/376>`_
 - Removes pytest-trio from the test dependencies `#620 <https://github.com/pytest-dev/pytest-asyncio/pull/620>`_
 
 0.22.0 (2023-10-31)

docs/source/reference/functions.rst

@@ -0,0 +1,9 @@
+=========
+Functions
+=========
+
+is_async_test
+=============
+Returns whether a specific pytest Item is an asynchronous test managed by pytest-asyncio.
+
+This function is intended to be used in pytest hooks or by plugins that depend on pytest-asyncio.

docs/source/reference/index.rst

@@ -7,6 +7,7 @@ Reference
 
   configuration
   fixtures/index
+  functions
   markers/index
   decorators/index
   changelog

pytest_asyncio/__init__.py

@@ -1,5 +1,5 @@
 """The main point for importing pytest-asyncio items."""
 from ._version import version as __version__  # noqa
-from .plugin import fixture
+from .plugin import fixture, is_async_test
 
-__all__ = ("fixture",)
+__all__ = ("fixture", "is_async_test")

pytest_asyncio/plugin.py

@@ -963,6 +963,11 @@ def event_loop_policy() -> AbstractEventLoopPolicy:
     return asyncio.get_event_loop_policy()
 
 
+def is_async_test(item: Item) -> bool:
+    """Returns whether a test item is a pytest-asyncio test"""
+    return isinstance(item, PytestAsyncioFunction)
+
+
 def _unused_port(socket_type: int) -> int:
     """Find an unused localhost port from 1024-65535 and return it."""
     with contextlib.closing(socket.socket(type=socket_type)) as sock:

tests/test_is_async_test.py

@@ -0,0 +1,100 @@
+from textwrap import dedent
+
+from pytest import Pytester
+
+
+def test_returns_false_for_sync_item(pytester: Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import pytest
+            import pytest_asyncio
+
+            def test_sync():
+                pass
+
+            def pytest_collection_modifyitems(items):
+                async_tests = [
+                    item
+                    for item in items
+                    if pytest_asyncio.is_async_test(item)
+                ]
+                assert len(async_tests) == 0
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_returns_true_for_marked_coroutine_item_in_strict_mode(pytester: Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import pytest
+            import pytest_asyncio
+
+            @pytest.mark.asyncio
+            async def test_coro():
+                pass
+
+            def pytest_collection_modifyitems(items):
+                async_tests = [
+                    item
+                    for item in items
+                    if pytest_asyncio.is_async_test(item)
+                ]
+                assert len(async_tests) == 1
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_returns_false_for_unmarked_coroutine_item_in_strict_mode(pytester: Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import pytest
+            import pytest_asyncio
+
+            async def test_coro():
+                pass
+
+            def pytest_collection_modifyitems(items):
+                async_tests = [
+                    item
+                    for item in items
+                    if pytest_asyncio.is_async_test(item)
+                ]
+                assert len(async_tests) == 0
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(skipped=1)
+
+
+def test_returns_true_for_unmarked_coroutine_item_in_auto_mode(pytester: Pytester):
+    pytester.makepyfile(
+        dedent(
+            """\
+            import pytest
+            import pytest_asyncio
+
+            async def test_coro():
+                pass
+
+            def pytest_collection_modifyitems(items):
+                async_tests = [
+                    item
+                    for item in items
+                    if pytest_asyncio.is_async_test(item)
+                ]
+                assert len(async_tests) == 1
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=auto")
+    result.assert_outcomes(passed=1)