[feat] Added the asyncio_default_fixture_loop_scope configuration option.

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

Author: seifertm

docs/source/reference/configuration.rst

@@ -2,6 +2,12 @@
 Configuration
 =============
 
+asyncio_default_fixture_loop_scope
+==================================
+Determines the default event loop scope of asynchronous fixtures. When this configuration option is unset, it defaults to the fixture scope. In future versions of pytest-asyncio, the value will default to ``function`` when unset. Possible values are: ``function``, ``class``, ``module``, ``package``, ``session``
+
+asyncio_mode
+============
 The pytest-asyncio mode can be set by the ``asyncio_mode`` configuration option in the `configuration file
 <https://docs.pytest.org/en/latest/reference/customize.html>`_:
 

pytest_asyncio/plugin.py

@@ -103,6 +103,12 @@ def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None
         help="default value for --asyncio-mode",
         default="strict",
     )
+    parser.addini(
+        "asyncio_default_fixture_loop_scope",
+        type="string",
+        help="default scope of the asyncio event loop used to execute async fixtures",
+        default=None,
+    )
 
 
 @overload
@@ -189,8 +195,20 @@ def _get_asyncio_mode(config: Config) -> Mode:
         )
 
 
+_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET = """\
+The configuration option "asyncio_default_fixture_loop_scope" is unset.
+The event loop scope for asynchronous fixtures will default to the fixture caching \
+scope. Future versions of pytest-asyncio will default the loop scope for asynchronous \
+fixtures to function scope. Set the default fixture loop scope explicitly in order to \
+avoid unexpected behavior in the future. Valid fixture loop scopes are: \
+"function", "class", "module", "package", "session"
+"""
+
+
 def pytest_configure(config: Config) -> None:
-    """Inject documentation."""
+    default_loop_scope = config.getini("asyncio_default_fixture_loop_scope")
+    if not default_loop_scope:
+        warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET))
     config.addinivalue_line(
         "markers",
         "asyncio: "
@@ -203,14 +221,16 @@ def pytest_configure(config: Config) -> None:
 def pytest_report_header(config: Config) -> List[str]:
     """Add asyncio config to pytest header."""
     mode = _get_asyncio_mode(config)
-    return [f"asyncio: mode={mode}"]
+    default_loop_scope = config.getini("asyncio_default_fixture_loop_scope")
+    return [f"asyncio: mode={mode}, default_loop_scope={default_loop_scope}"]
 
 
 def _preprocess_async_fixtures(
     collector: Collector,
     processed_fixturedefs: Set[FixtureDef],
 ) -> None:
     config = collector.config
+    default_loop_scope = config.getini("asyncio_default_fixture_loop_scope")
     asyncio_mode = _get_asyncio_mode(config)
     fixturemanager = config.pluginmanager.get_plugin("funcmanage")
     assert fixturemanager is not None
@@ -225,7 +245,11 @@ def _preprocess_async_fixtures(
                 # Ignore async fixtures without explicit asyncio mark in strict mode
                 # This applies to pytest_trio fixtures, for example
                 continue
-            scope = getattr(func, "_loop_scope", None) or fixturedef.scope
+            scope = (
+                getattr(func, "_loop_scope", None)
+                or default_loop_scope
+                or fixturedef.scope
+            )
             if scope == "function":
                 event_loop_fixture_id: Optional[str] = "event_loop"
             else:

tests/test_fixture_loop_scopes.py

@@ -34,3 +34,103 @@ async def test_runs_in_same_loop_as_fixture(fixture):
     )
     result = pytester.runpytest_subprocess("--asyncio-mode=strict")
     result.assert_outcomes(passed=1)
+
+
+@pytest.mark.parametrize("default_loop_scope", ("function", "module", "session"))
+def test_default_loop_scope_config_option_changes_fixture_loop_scope(
+    pytester: Pytester,
+    default_loop_scope: str,
+):
+    pytester.makeini(
+        dedent(
+            f"""\
+            [pytest]
+            asyncio_default_fixture_loop_scope = {default_loop_scope}
+            """
+        )
+    )
+    pytester.makepyfile(
+        dedent(
+            f"""\
+            import asyncio
+            import pytest
+            import pytest_asyncio
+
+            @pytest_asyncio.fixture
+            async def fixture_loop():
+                return asyncio.get_running_loop()
+
+            @pytest.mark.asyncio(scope="{default_loop_scope}")
+            async def test_runs_in_fixture_loop(fixture_loop):
+                assert asyncio.get_running_loop() is fixture_loop
+            """
+        )
+    )
+    result = pytester.runpytest_subprocess("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_default_class_loop_scope_config_option_changes_fixture_loop_scope(
+    pytester: Pytester,
+):
+    pytester.makeini(
+        dedent(
+            """\
+            [pytest]
+            asyncio_default_fixture_loop_scope = class
+            """
+        )
+    )
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+            import pytest_asyncio
+
+            class TestClass:
+                @pytest_asyncio.fixture
+                async def fixture_loop(self):
+                    return asyncio.get_running_loop()
+
+                @pytest.mark.asyncio(scope="class")
+                async def test_runs_in_fixture_loop(self, fixture_loop):
+                    assert asyncio.get_running_loop() is fixture_loop
+            """
+        )
+    )
+    result = pytester.runpytest_subprocess("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_default_package_loop_scope_config_option_changes_fixture_loop_scope(
+    pytester: Pytester,
+):
+    pytester.makeini(
+        dedent(
+            """\
+            [pytest]
+            asyncio_default_fixture_loop_scope = package
+            """
+        )
+    )
+    pytester.makepyfile(
+        __init__="",
+        test_a=dedent(
+            """\
+            import asyncio
+            import pytest
+            import pytest_asyncio
+
+            @pytest_asyncio.fixture
+            async def fixture_loop():
+                return asyncio.get_running_loop()
+
+            @pytest.mark.asyncio(scope="package")
+            async def test_runs_in_fixture_loop(fixture_loop):
+                assert asyncio.get_running_loop() is fixture_loop
+            """
+        ),
+    )
+    result = pytester.runpytest_subprocess("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)