Remove use of deprecated function from test synchronizer

pytest_asyncio/plugin.py

@@ -437,6 +437,32 @@ def _can_substitute(item: Function) -> bool:
         """Returns whether the specified function can be replaced by this class"""
         raise NotImplementedError()
 
+    def runtest(self) -> None:
+        marker = self.get_closest_marker("asyncio")
+        assert marker is not None
+        default_loop_scope = _get_default_test_loop_scope(self.config)
+        loop_scope = _get_marked_loop_scope(marker, default_loop_scope)
+        runner_fixture_id = f"_{loop_scope}_scoped_runner"
+        runner = self._request.getfixturevalue(runner_fixture_id)
+        context = contextvars.copy_context()
+        synchronized_obj = _synchronize_coroutine(
+            getattr(*self._synchronization_target_attr), runner, context
+        )
+        with MonkeyPatch.context() as c:
+            c.setattr(*self._synchronization_target_attr, synchronized_obj)
+            super().runtest()
+
+    @property
+    def _synchronization_target_attr(self) -> tuple[object, str]:
+        """
+        Return the coroutine that needs to be synchronized during the test run.
+
+        This method is inteded to be overwritten by subclasses when they need to apply
+        the coroutine synchronizer to a value that's different from self.obj
+        e.g. the AsyncHypothesisTest subclass.
+        """
+        return self, "obj"
+
 
 class Coroutine(PytestAsyncioFunction):
     """Pytest item created by a coroutine"""
@@ -446,12 +472,6 @@ def _can_substitute(item: Function) -> bool:
         func = item.obj
         return inspect.iscoroutinefunction(func)
 
-    def runtest(self) -> None:
-        synchronized_obj = wrap_in_sync(self.obj)
-        with MonkeyPatch.context() as c:
-            c.setattr(self, "obj", synchronized_obj)
-            super().runtest()
-
 
 class AsyncGenerator(PytestAsyncioFunction):
     """Pytest item created by an asynchronous generator"""
@@ -488,12 +508,6 @@ def _can_substitute(item: Function) -> bool:
             func.__func__
         )
 
-    def runtest(self) -> None:
-        synchronized_obj = wrap_in_sync(self.obj)
-        with MonkeyPatch.context() as c:
-            c.setattr(self, "obj", synchronized_obj)
-            super().runtest()
-
 
 class AsyncHypothesisTest(PytestAsyncioFunction):
     """
@@ -510,11 +524,9 @@ def _can_substitute(item: Function) -> bool:
             and inspect.iscoroutinefunction(func.hypothesis.inner_test)
         )
 
-    def runtest(self) -> None:
-        synchronized_obj = wrap_in_sync(self.obj.hypothesis.inner_test)
-        with MonkeyPatch.context() as c:
-            c.setattr(self.obj.hypothesis, "inner_test", synchronized_obj)
-            super().runtest()
+    @property
+    def _synchronization_target_attr(self) -> tuple[object, str]:
+        return self.obj.hypothesis, "inner_test"
 
 
 # The function name needs to start with "pytest_"
@@ -652,28 +664,20 @@ def pytest_pyfunc_call(pyfuncitem: Function) -> object | None:
     return None
 
 
-def wrap_in_sync(
-    func: Callable[..., Awaitable[Any]],
+def _synchronize_coroutine(
+    func: Callable[..., CoroutineType],
+    runner: asyncio.Runner,
+    context: contextvars.Context,
 ):
     """
-    Return a sync wrapper around an async function executing it in the
-    current event loop.
+    Return a sync wrapper around a coroutine executing it in the
+    specified runner and context.
     """
 
     @functools.wraps(func)
     def inner(*args, **kwargs):
         coro = func(*args, **kwargs)
-        _loop = _get_event_loop_no_warn()
-        task = asyncio.ensure_future(coro, loop=_loop)
-        try:
-            _loop.run_until_complete(task)
-        except BaseException:
-            # run_until_complete doesn't get the result from exceptions
-            # that are not subclasses of `Exception`. Consume all
-            # exceptions to prevent asyncio's warning from logging.
-            if task.done() and not task.cancelled():
-                task.exception()
-            raise
+        runner.run(coro, context=context)
 
     return inner
 

tests/async_fixtures/test_async_fixtures_contextvars.py

@@ -6,7 +6,9 @@
 from __future__ import annotations
 
 from textwrap import dedent
+from typing import Literal
 
+import pytest
 from pytest import Pytester
 
 _prelude = dedent(
@@ -213,3 +215,56 @@ async def test(same_var_fixture):
     )
     result = pytester.runpytest("--asyncio-mode=strict")
     result.assert_outcomes(passed=1)
+
+
+def test_no_isolation_against_context_changes_in_sync_tests(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """
+            import pytest
+            import pytest_asyncio
+            from contextvars import ContextVar
+
+            _context_var = ContextVar("my_var")
+
+            def test_sync():
+                _context_var.set("new_value")
+
+            @pytest.mark.asyncio
+            async def test_async():
+                assert _context_var.get() == "new_value"
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)
+
+
+@pytest.mark.parametrize("loop_scope", ("function", "module"))
+def test_isolation_against_context_changes_in_async_tests(
+    pytester: Pytester, loop_scope: Literal["function", "module"]
+):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            f"""
+            import pytest
+            import pytest_asyncio
+            from contextvars import ContextVar
+
+            _context_var = ContextVar("my_var")
+
+            @pytest.mark.asyncio(loop_scope="{loop_scope}")
+            async def test_async_first():
+                _context_var.set("new_value")
+
+            @pytest.mark.asyncio(loop_scope="{loop_scope}")
+            async def test_async_second():
+                with pytest.raises(LookupError):
+                    _context_var.get()
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=2)