feat: apluggy integration

pyproject.toml

@@ -6,7 +6,8 @@ authors = ["Willem Thiart <himself@willemthiart.com>"]
 license = "MIT"
 
 [tool.poetry.dependencies]
-python = ">=3.7"
+python = ">=3.10"
+apluggy = "^0.9.7"
 
 [tool.poetry.dev-dependencies]
 

pytest_asyncio_cooperative/hookspecs.py

@@ -0,0 +1,71 @@
+from typing import Optional, Tuple
+
+from _pytest.fixtures import FixtureDef
+from _pytest.fixtures import SubRequest
+from _pytest.main import Session
+from _pytest.nodes import Item
+from _pytest.runner import call_and_report
+
+import apluggy as pluggy
+
+
+hookspec = pluggy.HookspecMarker('pytest-asyncio-cooperative')
+
+
+@hookspec(firstresult=True)
+async def pytest_runtestloop(session: "Session") -> Optional[object]:
+    ...
+
+
+@hookspec(firstresult=True)
+async def pytest_runtest_protocol(
+    item: "Item", nextitem: "Optional[Item]"
+) -> Optional[object]:
+    ...
+
+
+@hookspec
+async def pytest_runtest_logstart(
+    nodeid: str, location: Tuple[str, Optional[int], str]
+) -> None:
+    ...
+
+
+@hookspec
+async def pytest_runtest_logfinish(
+    nodeid: str, location: Tuple[str, Optional[int], str]
+) -> None:
+    ...
+
+
+@hookspec
+async def pytest_runtest_setup(item: "Item") -> None:
+    ...
+
+
+@hookspec
+async def pytest_runtest_call(item: "Item") -> None:
+    ...
+
+
+@hookspec
+async def pytest_runtest_teardown(item: "Item", nextitem: Optional["Item"]) -> None:
+    ...
+
+
+# -------------------------------------------------------------------------
+# Fixture related hooks
+# -------------------------------------------------------------------------
+
+
+@hookspec(firstresult=True)
+async def pytest_fixture_setup(
+    fixturedef: "FixtureDef[Any]", request: "SubRequest"
+) -> Optional[object]:
+    ...
+
+
+def pytest_fixture_post_finalizer(
+    fixturedef: "FixtureDef[Any]", request: "SubRequest"
+) -> None:
+    ...

pytest_asyncio_cooperative/plugin.py

@@ -5,13 +5,15 @@
 import time
 from sys import version_info as sys_version_info
 
+import apluggy as pluggy
+
 import pytest
 
-from _pytest.runner import call_and_report
 from _pytest.skipping import Skip, evaluate_skip_marks
 
 from .assertion import activate_assert_rewrite
 from .fixtures import fill_fixtures
+from . import hookspecs
 
 
 def pytest_addoption(parser):
@@ -50,6 +52,55 @@ def pytest_configure(config):
     )
 
 
+hookimpl = pluggy.HookimplMarker('pytest-asyncio-cooperative')
+
+
+class CorePlugin:
+    @hookimpl
+    async def pytest_runtest_call(self, item) -> None:
+        # Do setup
+        item.start_setup = time.time()
+        fixture_values, teardowns = await fill_fixtures(item)
+        item.stop_setup = time.time()
+
+        # This is a class method test so prepend `self`
+        if item.instance:
+            fixture_values.insert(0, item.instance)
+
+        async def do_teardowns():
+            item.start_teardown = time.time()
+            for teardown in teardowns:
+                if isinstance(teardown, collections.abc.Iterator):
+                    try:
+                        teardown.__next__()
+                    except StopIteration:
+                        pass
+                else:
+                    try:
+                        await teardown.__anext__()
+                    except StopAsyncIteration:
+                        pass
+            item.stop_teardown = time.time()
+
+        # Run test
+        item.start = time.time()
+        try:
+            await item.function(*fixture_values)
+        except:
+            # Teardown here otherwise we might leave fixtures with locks acquired
+            item.stop = time.time()
+            await do_teardowns()
+            raise
+
+        item.stop = time.time()
+        await do_teardowns()
+
+
+pm = pluggy.PluginManager('pytest-asyncio-cooperative')
+pm.add_hookspecs(hookspecs)
+pm.register(CorePlugin)
+
+
 @pytest.hookspec
 def pytest_runtest_makereport(item, call):
     # Tests are run outside of the normal place, so we have to inject our timings
@@ -82,42 +133,7 @@ def not_coroutine_failure(function_name: str, *args, **kwargs):
 
 
 async def test_wrapper(item):
-    # Do setup
-    item.start_setup = time.time()
-    fixture_values, teardowns = await fill_fixtures(item)
-    item.stop_setup = time.time()
-
-    # This is a class method test so prepend `self`
-    if item.instance:
-        fixture_values.insert(0, item.instance)
-
-    async def do_teardowns():
-        item.start_teardown = time.time()
-        for teardown in teardowns:
-            if isinstance(teardown, collections.abc.Iterator):
-                try:
-                    teardown.__next__()
-                except StopIteration:
-                    pass
-            else:
-                try:
-                    await teardown.__anext__()
-                except StopAsyncIteration:
-                    pass
-        item.stop_teardown = time.time()
-
-    # Run test
-    item.start = time.time()
-    try:
-        await item.function(*fixture_values)
-    except:
-        # Teardown here otherwise we might leave fixtures with locks acquired
-        item.stop = time.time()
-        await do_teardowns()
-        raise
-
-    item.stop = time.time()
-    await do_teardowns()
+    await pm.ahook.pytest_runtest_call(item=item)
 
 
 # TODO: move to hypothesis module