Support pytest_asyncio_cooperative-marked synchronous tests

Author: gerlero

pytest_asyncio_cooperative/plugin.py

@@ -73,14 +73,6 @@ def pytest_runtest_makereport(item, call):
             call.duration = call.stop - call.start
 
 
-def not_coroutine_failure(function_name: str, *args, **kwargs):
-    raise Exception(
-        f"Function {function_name} is not a coroutine.\n"
-        f"Tests with the `@pytest.mark.asyncio_cooperative` mark MUST be coroutines.\n"
-        f"Please add the `async` keyword to the test function."
-    )
-
-
 async def test_wrapper(item):
     # Do setup
     item.start_setup = time.time()
@@ -109,7 +101,10 @@ async def do_teardowns():
     # Run test
     item.start = time.time()
     try:
-        await item.function(*fixture_values)
+        if inspect.iscoroutinefunction(item.function):
+            await item.function(*fixture_values)
+        else:
+            item.function(*fixture_values)
     except:
         # Teardown here otherwise we might leave fixtures with locks acquired
         item.stop = time.time()
@@ -158,17 +153,11 @@ def async_to_sync(*args, **kwargs):
     item.stop_teardown = time.time()
 
 
-class NotCoroutine(Exception):
-    pass
-
-
 def item_to_task(item):
-    if inspect.iscoroutinefunction(item.function):
-        return test_wrapper(item)
-    elif getattr(item.function, "is_hypothesis_test", False):
+    if getattr(item.function, "is_hypothesis_test", False):
         return hypothesis_test_wrapper(item)
     else:
-        raise NotCoroutine
+        return test_wrapper(item)
 
 
 def _run_test_loop(tasks, session, run_tests):
@@ -221,12 +210,7 @@ def pytest_runtestloop(session):
 
         # Coerce into a task
         if "asyncio_cooperative" in markers:
-            try:
-                task = item_to_task(item)
-            except NotCoroutine:
-                item.runtest = functools.partial(not_coroutine_failure, item.name)
-                item.ihook.pytest_runtest_protocol(item=item, nextitem=None)
-                continue
+            task = item_to_task(item)
 
             item._flakey = "flakey" in markers
             item_by_coro[task] = item

tests/test_core.py

@@ -91,3 +91,32 @@ def test_b():
 
     # we expect this:
     result.assert_outcomes(failed=1, passed=1)
+
+
+def test_synchronous_test_with_async_fixture(testdir):
+    testdir.makepyfile(
+        """
+        import asyncio
+        import pytest
+
+
+        @pytest.fixture
+        async def async_fixture():
+            return await asyncio.sleep(1, 42)
+
+        @pytest.fixture
+        def sync_fixture():
+            return 42
+
+        @pytest.mark.asyncio_cooperative
+        async def test_async(async_fixture, sync_fixture):
+            assert async_fixture == sync_fixture == 42
+
+        @pytest.mark.asyncio_cooperative
+        def test_sync(async_fixture, sync_fixture):
+            assert async_fixture == sync_fixture == 42
+    """
+    )
+
+    result = testdir.runpytest()
+    result.assert_outcomes(passed=2)

tests/test_fail.py

@@ -1,34 +1,5 @@
 import pytest
 
-from .conftest import includes_lines
-
-
-def test_function_must_be_async(testdir):
-    testdir.makeconftest("""""")
-
-    testdir.makepyfile(
-        """
-        import asyncio
-        import pytest
-
-
-        @pytest.mark.asyncio_cooperative
-        def test_a():
-            assert 1 == 1
-    """
-    )
-
-    expected_lines = [
-        "E       Exception: Function test_a is not a coroutine.",
-        "E       Tests with the `@pytest.mark.asyncio_cooperative` mark MUST be coroutines.",
-        "E       Please add the `async` keyword to the test function.",
-    ]
-
-    result = testdir.runpytest()
-    assert includes_lines(expected_lines, result.stdout.lines)
-
-    result.assert_outcomes(failed=1)
-
 
 @pytest.mark.parametrize("dur1, dur2, expectedfails, expectedpasses", [
     (1.1, 2, 2, 0),