refactor: Deduplicate runtest logic for PytestAsyncioFunction subclasses.

Author: seifertm

pytest_asyncio/plugin.py

@@ -437,15 +437,6 @@ def _can_substitute(item: Function) -> bool:
         """Returns whether the specified function can be replaced by this class"""
         raise NotImplementedError()
 
-
-class Coroutine(PytestAsyncioFunction):
-    """Pytest item created by a coroutine"""
-
-    @staticmethod
-    def _can_substitute(item: Function) -> bool:
-        func = item.obj
-        return inspect.iscoroutinefunction(func)
-
     def runtest(self) -> None:
         marker = self.get_closest_marker("asyncio")
         assert marker is not None
@@ -454,11 +445,33 @@ def runtest(self) -> None:
         runner_fixture_id = f"_{loop_scope}_scoped_runner"
         runner = self._request.getfixturevalue(runner_fixture_id)
         context = contextvars.copy_context()
-        synchronized_obj = wrap_in_sync(self.obj, runner, context)
+        synchronized_obj = wrap_in_sync(
+            getattr(*self._synchronization_target_attr), runner, context
+        )
         with MonkeyPatch.context() as c:
-            c.setattr(self, "obj", synchronized_obj)
+            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"""
+
+    @staticmethod
+    def _can_substitute(item: Function) -> bool:
+        func = item.obj
+        return inspect.iscoroutinefunction(func)
+
 
 class AsyncGenerator(PytestAsyncioFunction):
     """Pytest item created by an asynchronous generator"""
@@ -495,19 +508,6 @@ def _can_substitute(item: Function) -> bool:
             func.__func__
         )
 
-    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 = wrap_in_sync(self.obj, runner, context=context)
-        with MonkeyPatch.context() as c:
-            c.setattr(self, "obj", synchronized_obj)
-            super().runtest()
-
 
 class AsyncHypothesisTest(PytestAsyncioFunction):
     """
@@ -524,18 +524,9 @@ def _can_substitute(item: Function) -> bool:
             and inspect.iscoroutinefunction(func.hypothesis.inner_test)
         )
 
-    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 = wrap_in_sync(self.obj.hypothesis.inner_test, runner, context)
-        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_"