Incomplete need to figure out how to get loop_factory / multiple into asyncio.Runner

Author: Vizonex

pytest_asyncio/plugin.py

@@ -372,6 +372,8 @@ def restore_contextvars():
 class PytestAsyncioFunction(Function):
     """Base class for all test functions managed by pytest-asyncio."""
 
+    loop_factory: Callable[[], AbstractEventLoop] | None
+
     @classmethod
     def item_subclass_for(cls, item: Function, /) -> type[PytestAsyncioFunction] | None:
         """
@@ -386,12 +388,18 @@ def item_subclass_for(cls, item: Function, /) -> type[PytestAsyncioFunction] | N
         return None
 
     @classmethod
-    def _from_function(cls, function: Function, /) -> Function:
+    def _from_function(
+        cls,
+        function: Function,
+        loop_factory: Callable[[], AbstractEventLoop] | None = None,
+        /,
+    ) -> Function:
         """
         Instantiates this specific PytestAsyncioFunction type from the specified
         Function item.
         """
         assert function.get_closest_marker("asyncio")
+
         subclass_instance = cls.from_parent(
             function.parent,
             name=function.name,
@@ -401,6 +409,7 @@ def _from_function(cls, function: Function, /) -> Function:
             keywords=function.keywords,
             originalname=function.originalname,
         )
+        subclass_instance.loop_factory = loop_factory
         subclass_instance.own_markers = function.own_markers
         assert subclass_instance.own_markers == function.own_markers
         return subclass_instance
@@ -525,9 +534,27 @@ def pytest_pycollect_makeitem_convert_async_functions_to_subclass(
                     node.config
                 ) == Mode.AUTO and not node.get_closest_marker("asyncio"):
                     node.add_marker("asyncio")
-                if node.get_closest_marker("asyncio"):
-                    updated_item = specialized_item_class._from_function(node)
-        updated_node_collection.append(updated_item)
+                if asyncio_marker := node.get_closest_marker("asyncio"):
+                    if loop_factory := asyncio_marker.kwargs.get("loop_factory", None):
+                        # multiply if loop_factory is an iterable object of factories
+                        if hasattr(loop_factory, "__iter__"):
+                            updated_item = [
+                                specialized_item_class._from_function(node, lf)
+                                for lf in loop_factory
+                            ]
+                        else:
+                            updated_item = specialized_item_class._from_function(
+                                node, loop_factory
+                            )
+                    else:
+                        updated_item = specialized_item_class._from_function(node)
+
+        # we could have multiple factroies to test if so,
+        # multiply the number of functions for us...
+        if isinstance(updated_item, list):
+            updated_node_collection.extend(updated_item)
+        else:
+            updated_node_collection.append(updated_item)
     hook_result.force_result(updated_node_collection)
 
 
@@ -546,20 +573,6 @@ def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[No
         _set_event_loop(old_loop)
 
 
-@contextlib.contextmanager
-def _temporary_event_loop(loop: AbstractEventLoop):
-    try:
-        old_event_loop = asyncio.get_event_loop()
-    except RuntimeError:
-        old_event_loop = None
-
-    asyncio.set_event_loop(old_event_loop)
-    try:
-        yield
-    finally:
-        asyncio.set_event_loop(old_event_loop)
-
-
 def _get_event_loop_policy() -> AbstractEventLoopPolicy:
     with warnings.catch_warnings():
         warnings.simplefilter("ignore", DeprecationWarning)
@@ -669,12 +682,15 @@ def pytest_runtest_setup(item: pytest.Item) -> None:
     marker = item.get_closest_marker("asyncio")
     if marker is None:
         return
+    getattr(marker, "loop_factory", None)
     default_loop_scope = _get_default_test_loop_scope(item.config)
     loop_scope = _get_marked_loop_scope(marker, default_loop_scope)
     runner_fixture_id = f"_{loop_scope}_scoped_runner"
-    fixturenames = item.fixturenames  # type: ignore[attr-defined]
+    fixturenames: list[str] = item.fixturenames  # type: ignore[attr-defined]
+
     if runner_fixture_id not in fixturenames:
         fixturenames.append(runner_fixture_id)
+
     obj = getattr(item, "obj", None)
     if not getattr(obj, "hypothesis", False) and getattr(
         obj, "is_hypothesis_test", False
@@ -701,8 +717,15 @@ def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None:
         or default_loop_scope
         or fixturedef.scope
     )
+    # XXX: Currently Confused as to where to debug and harvest and get the runner to use the loop_factory argument.
+    loop_factory = getattr(fixturedef.func, "loop_factory", None)
+
+    print(f"LOOP FACTORY: {loop_factory} {fixturedef.func}")
+    sys.stdout.flush()
+
     runner_fixture_id = f"_{loop_scope}_scoped_runner"
-    runner = request.getfixturevalue(runner_fixture_id)
+    runner: Runner = request.getfixturevalue(runner_fixture_id)
+
     synchronizer = _fixture_synchronizer(fixturedef, runner, request)
     _make_asyncio_fixture_function(synchronizer, loop_scope)
     with MonkeyPatch.context() as c:
@@ -727,9 +750,12 @@ def _get_marked_loop_scope(
 ) -> _ScopeName:
     assert asyncio_marker.name == "asyncio"
     if asyncio_marker.args or (
-        asyncio_marker.kwargs and set(asyncio_marker.kwargs) - {"loop_scope", "scope"}
+        asyncio_marker.kwargs
+        and set(asyncio_marker.kwargs) - {"loop_scope", "scope", "loop_factory"}
     ):
-        raise ValueError("mark.asyncio accepts only a keyword argument 'loop_scope'.")
+        raise ValueError(
+            "mark.asyncio accepts only keyword arguments 'loop_scope', 'loop_factory'."
+        )
     if "scope" in asyncio_marker.kwargs:
         if "loop_scope" in asyncio_marker.kwargs:
             raise pytest.UsageError(_DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR)
@@ -795,12 +821,6 @@ def _scoped_runner(
     )
 
 
-@pytest.fixture(scope="session", autouse=True)
-def new_event_loop() -> AbstractEventLoop:
-    """Creates a new eventloop for different tests being ran"""
-    return asyncio.new_event_loop()
-
-
 @pytest.fixture(scope="session", autouse=True)
 def event_loop_policy() -> AbstractEventLoopPolicy:
     """Return an instance of the policy used to create asyncio event loops."""

tests/markers/test_invalid_arguments.py

@@ -40,9 +40,7 @@ async def test_anything():
     )
     result = pytester.runpytest_subprocess()
     result.assert_outcomes(errors=1)
-    result.stdout.fnmatch_lines(
-        ["*ValueError: mark.asyncio accepts only a keyword argument*"]
-    )
+    result.stdout.fnmatch_lines([""])
 
 
 def test_error_when_wrong_keyword_argument_is_passed(
@@ -62,7 +60,9 @@ async def test_anything():
     result = pytester.runpytest_subprocess()
     result.assert_outcomes(errors=1)
     result.stdout.fnmatch_lines(
-        ["*ValueError: mark.asyncio accepts only a keyword argument 'loop_scope'*"]
+        [
+            "*ValueError: mark.asyncio accepts only keyword arguments 'loop_scope', 'loop_factory'*"
+        ]
     )
 
 
@@ -83,5 +83,7 @@ async def test_anything():
     result = pytester.runpytest_subprocess()
     result.assert_outcomes(errors=1)
     result.stdout.fnmatch_lines(
-        ["*ValueError: mark.asyncio accepts only a keyword argument*"]
+        [
+            "*ValueError: mark.asyncio accepts only keyword arguments 'loop_scope', 'loop_factory'*"
+        ]
     )

tests/test_asyncio_mark.py

@@ -223,3 +223,35 @@ async def test_a(session_loop_fixture):
 
     result = pytester.runpytest("--asyncio-mode=auto")
     result.assert_outcomes(passed=1)
+
+
+def test_asyncio_marker_event_loop_factories(pytester: Pytester):
+    pytester.makeini(
+        dedent(
+            """\
+            [pytest]
+            asyncio_default_fixture_loop_scope = function
+            asyncio_default_test_loop_scope = module
+            """
+        )
+    )
+
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            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"
+            """
+        )
+    )
+
+    result = pytester.runpytest("--asyncio-mode=auto")
+    result.assert_outcomes(passed=1)