wip: Add tests for additional edge cases and try to cover them.

Author: seifertm

pytest_asyncio/plugin.py

@@ -836,14 +836,28 @@ def _get_default_test_loop_scope(config: Config) -> Any:
 def _get_loop_facotry(
     request: FixtureRequest,
 ) -> Callable[[], AbstractEventLoop] | None:
-    if asyncio_mark := request._pyfuncitem.get_closest_marker("asyncio"):
+    loop_factories = []
+    asyncio_mark = request._pyfuncitem.get_closest_marker("asyncio")
+    if asyncio_mark is not None:
         # The loop_factory is defined on an asyncio marker
         factory = asyncio_mark.kwargs.get("loop_factory", None)
-        return factory
-    else:
-        # The loop_factory is pulled in via a fixture
-        top_request = list(request._iter_chain())[-1]._parent_request
-        return top_request._pyfuncitem.__dict__.get("_loop_factory", None)  # type: ignore[attr-defined]
+        loop_factories.append(factory)
+    # The loop_factory is defined in a transitive fixture
+    current_request = request
+    for r in request._iter_chain():
+        current_request = r
+        loop_factory = getattr(current_request._fixturedef.func, "_loop_factory", None)
+        loop_factories.append(loop_factory)
+    defined_loop_factories = [factory for factory in loop_factories if factory] or [
+        None
+    ]
+    print(defined_loop_factories)
+    if len(defined_loop_factories) > 1:
+        print(defined_loop_factories)
+        raise pytest.UsageError(
+            "Multiple loop factories defined for {request.scope}-scoped loop."
+        )
+    return defined_loop_factories[0]
 
 
 def _create_scoped_runner_fixture(scope: _ScopeName) -> Callable:

tests/async_fixtures/test_async_fixtures_contextvars.py

@@ -101,7 +101,7 @@ def test(check_var_fixture):
         """
         )
     )
-    result = pytester.runpytest("--asyncio-mode=strict")
+    result = pytester.runpytest("--asyncio-mode=strict", "-s")
     result.assert_outcomes(passed=1)
 
 

tests/test_asyncio_mark.py

@@ -148,15 +148,15 @@ async def test_a():
     )
 
 
-def test_asyncio_marker_fallbacks_to_configured_default_loop_scope_if_not_set(
+def test_asyncio_marker_uses_marker_loop_scope_even_if_config_is_set(
     pytester: Pytester,
 ):
     pytester.makeini(
         dedent(
             """\
             [pytest]
             asyncio_default_fixture_loop_scope = function
-            asyncio_default_test_loop_scope = session
+            asyncio_default_test_loop_scope = module
             """
         )
     )
@@ -175,6 +175,7 @@ async def session_loop_fixture():
                 global loop
                 loop = asyncio.get_running_loop()
 
+            @pytest.mark.asyncio(loop_scope="session")
             async def test_a(session_loop_fixture):
                 global loop
                 assert asyncio.get_running_loop() is loop
@@ -186,56 +187,125 @@ async def test_a(session_loop_fixture):
     result.assert_outcomes(passed=1)
 
 
-def test_asyncio_marker_uses_marker_loop_scope_even_if_config_is_set(
-    pytester: Pytester,
-):
-    pytester.makeini(
+def test_uses_loop_factory_from_test(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
         dedent(
             """\
-            [pytest]
-            asyncio_default_fixture_loop_scope = function
-            asyncio_default_test_loop_scope = module
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module")
+            async def any_fixture():
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+
+            @pytest.mark.asyncio(loop_scope="module", loop_factory=CustomEventLoop)
+            async def test_set_loop_factory(any_fixture):
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
             """
         )
     )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
 
+def test_uses_loop_factory_from_fixture(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
     pytester.makepyfile(
         dedent(
             """\
             import asyncio
             import pytest_asyncio
             import pytest
 
-            loop: asyncio.AbstractEventLoop
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
 
-            @pytest_asyncio.fixture(loop_scope="session", scope="session")
-            async def session_loop_fixture():
-                global loop
-                loop = asyncio.get_running_loop()
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def any_fixture():
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
 
-            @pytest.mark.asyncio(loop_scope="session")
-            async def test_a(session_loop_fixture):
-                global loop
-                assert asyncio.get_running_loop() is loop
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_set_loop_factory(any_fixture):
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
             """
         )
     )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
 
-    result = pytester.runpytest("--asyncio-mode=auto")
+
+def test_uses_loop_factory_from_transitive_fixture(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def transitive_fixture():
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+
+            @pytest_asyncio.fixture(loop_scope="module")
+            async def any_fixture(transitive_fixture):
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_set_loop_factory(any_fixture):
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
     result.assert_outcomes(passed=1)
 
 
-def test_asyncio_marker_event_loop_factories(pytester: Pytester):
-    pytester.makeini(
+def test_conflicting_loop_factories_in_tests_raise_error(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
         dedent(
             """\
-            [pytest]
-            asyncio_default_fixture_loop_scope = function
-            asyncio_default_test_loop_scope = module
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            class AnotherCustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest.mark.asyncio(loop_scope="module", loop_factory=CustomEventLoop)
+            async def test_with_custom_loop_factory():
+                ...
+
+            @pytest.mark.asyncio(
+                loop_scope="module",
+                loop_factory=AnotherCustomEventLoop
+            )
+            async def test_with_a_different_custom_loop_factory():
+                ...
             """
         )
     )
 
+    result = pytester.runpytest("--asyncio-mode=strict", "-s", "--setup-show")
+    result.assert_outcomes(errors=2)
+
+
+def test_conflicting_loop_factories_in_tests_and_fixtures_raise_error(
+    pytester: Pytester,
+):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
     pytester.makepyfile(
         dedent(
             """\
@@ -246,22 +316,109 @@ def test_asyncio_marker_event_loop_factories(pytester: Pytester):
             class CustomEventLoop(asyncio.SelectorEventLoop):
                 pass
 
-            @pytest.mark.asyncio(loop_factory=CustomEventLoop)
-            async def test_has_different_event_loop():
-                assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoop"
+            class AnotherCustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def fixture_with_custom_loop_factory():
+                ...
+
+            @pytest.mark.asyncio(
+                loop_scope="module",
+                loop_factory=AnotherCustomEventLoop
+            )
+            async def test_trying_to_override_fixtures_loop_factory(
+                fixture_with_custom_loop_factory
+            ):
+                # Fails, because it tries to use a different loop factory on the
+                # same runner as the first test
+                ...
+            """
+        )
+    )
+
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1, errors=1)
 
-            @pytest_asyncio.fixture(loop_factory=CustomEventLoop)
-            async def custom_fixture():
-                yield asyncio.get_running_loop()
 
-            async def test_with_fixture(custom_fixture):
-                # Both of these should be the same...
-                type(asyncio.get_running_loop()).__name__ == "CustomEventLoop"
-                type(custom_fixture).__name__ == "CustomEventLoop"
+def test_conflicting_loop_factories_in_fixtures_raise_error(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
 
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            class AnotherCustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def fixture_with_custom_loop_factory():
+                ...
+
+            @pytest_asyncio.fixture(
+                loop_scope="module",
+                loop_factory=AnotherCustomEventLoop
+            )
+            async def another_fixture_with_custom_loop_factory():
+                ...
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_requesting_two_fixtures_with_different_loop_facoties(
+                fixture_with_custom_loop_factory,
+                another_fixture_with_custom_loop_factory,
+            ):
+                ...
             """
         )
     )
 
-    result = pytester.runpytest("--asyncio-mode=auto")
-    result.assert_outcomes(passed=2)
+    result = pytester.runpytest("--asyncio-mode=strict", "-s", "--setup-show")
+    result.assert_outcomes(errors=1)
+
+
+def test_conflicting_loop_factories_in_transitive_fixtures_raise_error(
+    pytester: Pytester,
+):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            class AnotherCustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def fixture_with_custom_loop_factory():
+                ...
+
+            @pytest_asyncio.fixture(
+                loop_scope="module",
+                loop_factory=AnotherCustomEventLoop
+            )
+            async def another_fixture_with_custom_loop_factory(
+                fixture_with_custom_loop_factory
+            ):
+                ...
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_requesting_two_fixtures_with_different_loop_facories(
+                another_fixture_with_custom_loop_factory,
+            ):
+                ...
+            """
+        )
+    )
+
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(errors=1)