fix: handle session fixtures with exceptions (#81)

* fix: handle session fixtures with exceptions
* test: update tests

Author: willemt

pytest_asyncio_cooperative/fixtures.py

@@ -1,5 +1,6 @@
 import asyncio
 import inspect
+import types
 from typing import List
 from typing import Union
 
@@ -114,6 +115,11 @@ def __init__(self, wrapped_func):
         self.lock = asyncio.Lock()
         self.wrapped_func = wrapped_func
 
+        # Trying to fool pytest's use of inspect.isgeneratorfunction
+        self.__defaults__ = getattr(wrapped_func, "__defaults__", None)
+        self.__kwdefaults__ = getattr(wrapped_func, "__kwdefaults__", None)
+        self.__annotations__ = getattr(wrapped_func, "__annotations__", {})
+
     @property
     def __code__(self):
         return self.wrapped_func.__code__

tests/test_fixture.py

@@ -1,3 +1,5 @@
+import sys
+
 import pytest
 
 
@@ -463,6 +465,7 @@ async def test_a(aaa, bbb):
     result.assert_outcomes(passed=1)
 
 
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="Requires Python 3.10+")
 @pytest.mark.parametrize("scope", ["module", "session"])
 @pytest.mark.parametrize("def_", ["def", "async def"])
 @pytest.mark.parametrize("ret", ["return", "yield"])
@@ -496,6 +499,51 @@ async def test_b(shared_fixture):
 
     result = testdir.runpytest()
 
+    if fail:
+        if ret == "yield" and def_ == "def":
+            result.assert_outcomes(errors=2)
+        else:
+            result.assert_outcomes(failed=2)
+        # Should be errors instead of failures
+        # https://github.com/willemt/pytest-asyncio-cooperative/issues/42
+    else:
+        result.assert_outcomes(passed=2)
+
+
+@pytest.mark.skipif(sys.version_info > (3, 9), reason="Legacy behaviour")
+@pytest.mark.parametrize("scope", ["module", "session"])
+@pytest.mark.parametrize("def_", ["def", "async def"])
+@pytest.mark.parametrize("ret", ["return", "yield"])
+@pytest.mark.parametrize("fail", [False, True])
+def test_shared_fixture_caching_pre_py3_10(testdir, scope, def_, ret, fail):
+    testdir.makepyfile(
+        f"""
+        import pytest
+        import time
+
+        called = False
+        @pytest.fixture(scope="{scope}")
+        {def_} shared_fixture():
+            global called
+            if called:
+                assert {fail}
+            else:
+                called = True
+                assert not {fail}
+            {ret}
+
+        @pytest.mark.asyncio_cooperative
+        async def test_a(shared_fixture):
+            assert True
+
+        @pytest.mark.asyncio_cooperative
+        async def test_b(shared_fixture):
+            assert True
+    """
+    )
+
+    result = testdir.runpytest()
+
     if fail:
         result.assert_outcomes(failed=2)
         # Should be errors instead of failures

tests/test_fixture_ducktyping.py

@@ -0,0 +1,16 @@
+import inspect
+import sys
+
+import pytest
+
+from pytest_asyncio_cooperative.fixtures import CachedGen
+
+
+def my_function():
+    yield "Hello, World!"
+
+
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="Requires Python 3.10+")
+def test_isgeneratorfunction_duck_typing():
+    wrapped = CachedGen(my_function)
+    assert inspect.isgeneratorfunction(wrapped)

tests/test_fixture_session.py

@@ -0,0 +1,44 @@
+import sys
+
+import pytest
+
+
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="Requires Python 3.10+")
+def test_session_fixture_with_exception(testdir):
+    testdir.makepyfile(
+        """
+        import asyncio
+
+        import pytest
+
+
+        class Foo:
+            def reset(self):
+                print("Foo reset")
+
+
+        @pytest.fixture(scope="session")
+        def session():
+            raise NotImplementedError("xxx")
+            foo = Foo()
+            yield foo
+
+
+        @pytest.fixture
+        def regularfixture(session):
+            session.reset()
+            yield session
+
+
+        @pytest.mark.asyncio_cooperative
+        async def test_1(regularfixture):
+            await asyncio.sleep(1)
+
+    """
+    )
+
+    result = testdir.runpytest("-q")
+
+    assert "object has no attribute 'reset'" not in "".join(result.outlines)
+
+    result.assert_outcomes(passed=0, failed=0, errors=1)