Repository: delfick/alt-pytest-asyncio (language: python)

Issue:
Using pytest markers to associate timeouts with fixtures is not officially supported by pytest.
Our tests rely on configuring fixture timeouts through markers. This currently causes warnings in pytest 8.3.3, and will likely break in future versions when the warning becomes an error.

See the example test below:
```
import pytest
import asyncio

@pytest.fixture()
@pytest.mark.async_timeout(0.1)
async def my_amazing_fixture():
    try:
        await asyncio.sleep(1)
        yield 1
    finally:
        await asyncio.sleep(1)


async def test_my_amazing_fixture(my_amazing_fixture):
    assert True
```

See also https://github.com/pytest-dev/pytest/issues/12850


Discussion / hints:

Fix patch:
diff --git a/README.rst b/README.rst
index 71f0999..e47a5ba 100644
--- a/README.rst
+++ b/README.rst
@@ -21,6 +21,20 @@ generator fixtures.
 Changelog
 ---------
 
+0.9.0 - TBD
+    * Enabling the plugin must now be done by adding ``alt_pytest_asyncio.enable``
+      to the pytest list of enabled plugins if it's not being manually enabled.
+    * Removed ``pytest.mark.async_timeout`` and replaced the functionality with
+      a fixture.
+    * Changed the exported api of the plugin
+
+        * ``alt_pytest_asyncio.plugin.OverrideLoop`` is now ``alt_pytest_asyncio.Loop``
+        * ``alt_pytest_asyncio.plugin.AltPytestAsyncioPlugin`` now takes ``managed_loop``
+          as a keyword argument instead of the first positional argument with the
+          name ``loop``.
+        * The new ``Loop`` that replaces ``OverrideLoop`` now has an attributed
+          ``controlled_loop`` instead of ``loop``.
+
 0.8.2 - 12 October 2024
     * Added type annotations
     * CI now tests against python 3.13
@@ -77,6 +91,18 @@ Changelog
       pytest.main from an existing event loop. I decided to make this it's
       own module so I can have tests for this code.
 
+Installation
+------------
+
+Most users of this plugin won't need to manually construct the plugin as that's
+only required if you're doing funky things where you want to manually call
+``pytest.main`` (see next section).
+
+For this majority case, enabling the plugin requires:
+* The plugin be installed in the python environment
+* Adding ``alt_pytest_asyncio.enable`` to the list of pytest plugins that are
+  enabled.
+
 Running from your own event loop
 --------------------------------
 
@@ -85,7 +111,7 @@ do something like:
 
 .. code-block:: python
 
-   from alt_pytest_asyncio.plugin import AltPytestAsyncioPlugin, run_coro_as_main
+   import alt_pytest_asyncio
    import nest_asyncio
    import asyncio
    import pytest
@@ -93,7 +119,13 @@ do something like:
    async def my_tests():
       await do_some_setup_before_pytest()
 
-      plugins = [AltPytestAsyncioPlugin(loop)]
+      loop: asyncio.AbstractEventLoop = ...
+
+      plugins = [
+        alt_pytest_asyncio.plugin.AltPytestAsyncioPlugin(
+            managed_loop=loop
+        ),
+      ]
 
       try:
           code = pytest.main([], plugins=plugins)
@@ -112,53 +144,127 @@ do something like:
       loop = asyncio.get_event_loop()
       nest_asyncio.apply(loop)
 
-      run_coro_as_main(loop, my_tests())
+      alt_pytest_asyncio.run_coro_as_main(loop, my_tests())
 
 Note that if you don't need to run pytest from an existing event loop, you don't
-need to do anything other than have alt_pytest_asyncio installed in your
-environment and you'll be able to just use async keywords on your fixtures and
+need to do anything other than have ``alt_pytest_asyncio`` installed in your
+environment and ``alt_pytest_asyncio.enable`` in your pytest plugins list
+and you'll be able to just use async keywords on your fixtures and
 tests.
 
 Timeouts
 --------
 
-alt_pytest_asyncio registers a ``pytest.mark.async_timeout(seconds)`` mark which
-you can use to set a timeout for your test.
+.. note:: The ``pytest.mark.async_timeout(seconds)`` that existed before
+   version 0.9.0 no longer has an effect and has been replaced with the fixtures
+   as mentioned below
+
+This plugin can configure the timeout for any async fixture or test using the
+``async_timeout`` fixture or by creating a ``default_async_timeout`` fixture.
 
 For example:
 
 .. code-block:: python
 
    import pytest
+   import alt_pytest_asyncio
 
-   @pytest.mark.async_timeout(10)
-   async def test_something():
+   AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
+
+   async def test_something(async_timeout: AsyncTimeout) -> None:
+      async_timeout.set_timeout_seconds(10)
       await something_that_may_take_a_while()
 
 This test will be cancelled after 10 seconds and raise an assertion error saying
 the test took too long and the file and line number where the test is.
 
-You can also use the async_timeout mark on coroutine fixtures:
+.. note:: The async_timeout passed into a fixture or a test is a new instance
+   specific to that fixture or test. Setting it in a fixture only affects that
+   fixture and setting it in a test only affects that test.
+
+You can also set a ``default_async_timeout`` fixture to change the default:
 
 .. code-block:: python
 
    import pytest
+   import alt_pytest_asyncio
+
+   AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
+
 
    @pytest.fixture()
-   @pytest.mark.async_timeout(0.5)
-   async def my_amazing_fixture():
+   def default_async_timeout() -> float:
+       return 0.5
+
+   @pytest.fixture
+   async def my_amazing_fixture() -> int:
+      # Will timeout because of our default 0.5
+      await asyncio.sleep(1)
+      return 1
+
+   @pytest.fixture
+   async def my_amazing_fixture(async_timeout: AsyncTimeout) -> int:
+      # Change timeout for just this fixture
+      async_timeout.set_timeout_seconds(2)
       await asyncio.sleep(1)
       return 1
 
+For fixtures that have a non function scope, they require a
+``{scope}_default_async_timeout`` fixture:
+
+.. code-block:: python
+
+   import pytest
+
+
+   @pytest.fixture(scope="session")
+   def session_default_async_timeout() -> float:
+       return 5
+
+   @pytest.fixture(scope="session")
+   async def some_fixture() -> None:
+       # timeout here is 5
+       pass
+
+   class TestStuff:
+       @pytest.fixture(scope="class")
+       async def some_fixture() -> None:
+           # timeout here is 5
+           pass
+
+       class TestMore:
+           @pytest.fixture(scope="class")
+           async def class_default_async_timeout() -> int:
+               return 8
+
+           @pytest.fixture(scope="class")
+           async def some_fixture() -> None:
+               # timeout here is 8
+               pass
+
+The plugin knows about the scopes ``function``, ``class``, ``module``, ``package``
+and ``session``. So say a ``scope="class"`` async fixture is executed, the closest
+``class_default_async_timeout`` fixture is used unless that doesn't exist, in which
+case ``module_default_async_timeout`` is used, otherwise ``package_default_async_timeout``,
+otherwise ``session_default_async_timeout``.
+
+There is a default ``session_default_async_timeout`` available which returns the
+value set by the ``default_async_timeout`` pytest option the plugin provides.
+
 And you can have a timeout on generator fixtures:
 
 .. code-block:: python
 
    import pytest
+   from collections.abc import Iterator
+   import alt_pytest_asyncio
+
+   AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
 
    @pytest.fixture()
-   @pytest.mark.async_timeout(0.5)
-   async def my_amazing_fixture():
+   async def my_amazing_fixture(async_timeout: AsyncTimeout) -> Iterator[int]:
+      async_timeout.set_timeout_seconds(0.5)
+
       try:
          await asyncio.sleep(1)
          yield 1
@@ -179,6 +285,15 @@ Note that if the timeout fires whilst you have the debugger active then the time
 will not cancel the current test. This is determined by checking if ``sys.gettrace()``
 returns a non-None value.
 
+The object that is provided when the fixture/test asks for ``async_timeout`` can
+be modified by overriding the ``async_timeout`` session scope'd fixture and
+returning an object that inherits from and implements
+``alt_pytest_asyncio.base.AsyncTimeoutProvider``. This is a python "abc" class
+with a single method ``load`` which is called to return the object given to the
+fixture or test. This object must implement
+``alt_pytest_asyncio.base.AsyncTimeout``. The default implementation can be found
+at ``alt_pytest_asyncio.plugin.LoadedAsyncTimeout``.
+
 Overriding the loop
 -------------------
 
@@ -191,26 +306,24 @@ then restore the original loop on exit.
 
 Usage looks like::
 
-    from alt_pytest_asyncio.plugin import OverrideLoop
+    import alt_pytest_asyncio
 
     class TestThing:
         @pytest.fixture(autouse=True)
-        def custom_loop(self):
-            with OverrideLoop() as custom_loop:
+        def custom_loop(self) -> alt_pytest_asyncio.protocols.Loop:
+            with alt_pytest_asyncio.Loop() as custom_loop:
                 yield custom_loop
 
-        def test_thing(self, custom_loop):
+        def test_thing(self, custom_loop: alt_pytest_asyncio.protocols.Loop):
             custom_loop.run_until_complete(my_thing())
 
 By putting the loop into an autouse fixture, all fixtures used by the test
 will have the custom loop. If you want to include module level fixtures too
 then use the OverrideLoop in a module level fixture too.
 
-OverrideLoop takes in a ``new_loop`` boolean that will make it so no new
-loop is set and asyncio is left with no default loop.
-
-The new loop itself (or None if new_loop is False) can be found in the
-``loop`` attribute of the object yielded by the context manager.
+If the Loop is instantiated with ``new_loop=True`` then it will create and manage
+a new event loop whilst it's being used as a context manager. This new loop
+will be available on the object as ``.controlled_loop``.
 
 The ``run_until_complete`` on the ``custom_loop`` in the above example will
 do a ``run_until_complete`` on the new loop, but in a way that means you
diff --git a/pyproject.toml b/pyproject.toml
index cc88314..b19d7bb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -21,7 +21,7 @@ dependencies = [
 ]
 
 [project.entry-points.pytest11]
-alt_pytest_asyncio = "alt_pytest_asyncio"
+alt_pytest_asyncio = "alt_pytest_asyncio.plugin"
 
 [project.urls]
 Homepage = "https://github.com/delfick/alt-pytest-asyncio"


Test patch:
diff --git a/alt_pytest_asyncio/__init__.py b/alt_pytest_asyncio/__init__.py
index 324218c..fe70a66 100644
--- a/alt_pytest_asyncio/__init__.py
+++ b/alt_pytest_asyncio/__init__.py
@@ -1,31 +1,6 @@
-"""pytest-cov: avoid already-imported warning: PYTEST_DONT_REWRITE."""
+from . import base, errors, plugin, protocols
+from .loop_manager import Loop
+from .machinery import run_coro_as_main
+from .version import VERSION
 
-import pytest
-
-from alt_pytest_asyncio.version import VERSION
-
-
-@pytest.hookimpl
-def pytest_addoption(parser: pytest.Parser) -> None:
-    """Add an option for default async timeouts"""
-    desc = "timeout in seconds before failing a test. This can be overriden with @pytest.mark.async_timeout(<my_timeout>)"
-
-    group = parser.getgroup(
-        "alt-pytest-asyncio", description="Alternative asyncio pytest plugin options"
-    )
-    group.addoption("--default-async-timeout", type=float, dest="default_async_timeout", help=desc)
-    parser.addini(
-        "default_async_timeout",
-        desc,
-        default=5,
-    )
-
-
-def pytest_configure(config: pytest.Config) -> None:
-    from alt_pytest_asyncio.plugin import AltPytestAsyncioPlugin
-
-    if not any(isinstance(p, AltPytestAsyncioPlugin) for p in config.pluginmanager.get_plugins()):
-        config.pluginmanager.register(AltPytestAsyncioPlugin())
-
-
-__all__ = ["pytest_addoption", "pytest_configure", "VERSION"]
+__all__ = ["plugin", "protocols", "errors", "base", "run_coro_as_main", "Loop", "VERSION"]
diff --git a/alt_pytest_asyncio/async_converters.py b/alt_pytest_asyncio/async_converters.py
deleted file mode 100644
index f01bf6d..0000000
--- a/alt_pytest_asyncio/async_converters.py
+++ /dev/null
@@ -1,394 +0,0 @@
-"""
-Helpers for converting async functions into sync functions.
-
-convert_fixtures and converted_async_test are used by the Plugin
-to convert async fixtures and async tests.
-
-We try our best to make sure error reporting is short and useful by doing some
-hacks with capturing errors inside run_until_complete rather than outside so
-that you don't get asyncio internals in the errors.
-"""
-
-import asyncio
-import contextvars
-import inspect
-import sys
-from collections.abc import AsyncGenerator, Awaitable, Callable, Coroutine, Generator
-from functools import wraps
-from typing import TYPE_CHECKING, Any, NoReturn, NotRequired, ParamSpec, TypedDict, TypeVar
-
-import pytest
-
-P_Args = ParamSpec("P_Args")
-T_Ret = TypeVar("T_Ret")
-
-
-class RunInfo(TypedDict):
-    e: NotRequired[BaseException]
-    func: NotRequired[Callable[..., object]]
-    ran_once: NotRequired[bool]
-    cancelled: bool
-
-
-class GenRunInfo(RunInfo):
-    gen_obj: AsyncGenerator[object]
-    finalizer: NotRequired[bool]
-
-
-def convert_fixtures(
-    ctx: contextvars.Context,
-    fixturedef: pytest.FixtureDef[object],
-    request: pytest.FixtureRequest,
-    node: pytest.Item,
-) -> None:
-    """Used to replace async fixtures"""
-    if not hasattr(fixturedef, "func"):
-        return
-
-    if hasattr(fixturedef.func, "__alt_asyncio_pytest_converted__"):
-        return
-
-    if inspect.iscoroutinefunction(fixturedef.func):
-        convert_async_coroutine_fixture(ctx, fixturedef, request, node)
-        fixturedef.func.__alt_asyncio_pytest_converted__ = True  # type: ignore[attr-defined]
-
-    elif inspect.isasyncgenfunction(fixturedef.func):
-        convert_async_gen_fixture(ctx, fixturedef, request, node)
-        fixturedef.func.__alt_asyncio_pytest_converted__ = True  # type: ignore[attr-defined]
-
-    elif inspect.isgeneratorfunction(fixturedef.func):
-        convert_sync_gen_fixture(ctx, fixturedef)
-        fixturedef.func.__alt_asyncio_pytest_converted__ = True  # type: ignore[attr-defined]
-
-    else:
-        convert_sync_fixture(ctx, fixturedef)
-        fixturedef.func.__alt_asyncio_pytest_converted__ = True  # type: ignore[attr-defined]
-
-
-def convert_sync_fixture(ctx: contextvars.Context, fixturedef: pytest.FixtureDef[object]) -> None:
-    """
-    Used to make sure a non-async fixture is run in our
-    asyncio contextvars
-    """
-    original: Callable[..., object] = fixturedef.func
-
-    @wraps(original)
-    def run_fixture(*args: object, **kwargs: object) -> object:
-        try:
-            ctx.run(lambda: None)
-
-            def run(func: Callable[..., object], *a: object, **kw: object) -> object:
-                return ctx.run(func, *a, **kw)
-        except RuntimeError:
-
-            def run(func: Callable[..., object], *a: object, **kw: object) -> object:
-                return func(*a, **kw)
-
-        return run(original, *args, **kwargs)
-
-    fixturedef.func = run_fixture  # type: ignore[misc]
-
-
-def convert_sync_gen_fixture(
-    ctx: contextvars.Context, fixturedef: pytest.FixtureDef[object]
-) -> None:
-    """
-    Used to make sure a non-async generator fixture is run in our
-    asyncio contextvars
-    """
-    _original: Any = fixturedef.func
-    original: Callable[..., Generator[object, None, None]] = _original
-
-    @wraps(original)
-    def run_fixture(*args: object, **kwargs: object) -> object:
-        try:
-            ctx.run(lambda: None)
-
-            def run(func: Callable[..., object], *a: object, **kw: object) -> object:
-                return ctx.run(func, *a, **kw)
-        except RuntimeError:
-
-            def run(func: Callable[..., object], *a: object, **kw: object) -> object:
-                return func(*a, **kw)
-
-        cm = original(*args, **kwargs)
-        value = run(cm.__next__)
-        try:
-            yield value
-            run(cm.__next__)
-        except StopIteration:
-            pass
-
-    fixturedef.func = run_fixture  # type: ignore[misc]
-
-
-def converted_async_test(
-    ctx: contextvars.Context,
-    test_tasks: dict[asyncio.AbstractEventLoop, list[asyncio.Task[object]]],
-    func: Callable[P_Args, Awaitable[T_Ret]],
-    timeout: float,
-    /,
-    *args: P_Args.args,
-    **kwargs: P_Args.kwargs,
-) -> T_Ret:
-    """Used to replace async tests"""
-    __tracebackhide__ = True
-
-    info: RunInfo = {"cancelled": False}
-    loop = asyncio.get_event_loop_policy().get_event_loop()
-
-    def look_at_task(t: asyncio.Task[object]) -> None:
-        test_tasks[loop].append(t)
-
-    info["func"] = func
-
-    return _run_and_raise(  # type: ignore[return-value]
-        ctx, loop, info, func, async_runner(func, timeout, info, args, kwargs), look_at_task
-    )
-
-
-def _find_async_timeout(func: Callable[..., object], node: pytest.Item) -> float:
-    timeout = float(
-        node.config.getoption("default_async_timeout", None)
-        or node.config.getini("default_async_timeout")
-    )
-
-    marker = node.get_closest_marker("async_timeout")
-    if marker:
-        timeout = marker.args[0]
-
-    if hasattr(func, "pytestmark"):
-        for m in func.pytestmark:
-            if m.name == "async_timeout":
-                timeout = m.args[0]
-
-    return timeout
-
-
-def _raise_maybe(func: Callable[..., object], info: RunInfo) -> None:
-    __tracebackhide__ = True
-
-    if hasattr(func, "__original__"):
-        func = func.__original__
-
-    fle = inspect.getfile(func)
-    if hasattr(func, "__func__"):
-        func = func.__func__
-    lineno = func.__code__.co_firstlineno
-
-    def raise_error() -> NoReturn:
-        if info["cancelled"]:
-            assert False, f"Took too long to complete: {fle}:{lineno}"
-        raise info["e"]
-
-    if info.get("e"):
-        # Use a separate function so when --tb=short is not set we don't get
-        # this entire function in the output
-        raise_error()
-
-
-def _run_and_raise(
-    ctx: contextvars.Context,
-    loop: asyncio.AbstractEventLoop,
-    info: RunInfo,
-    func: Callable[P_Args, T_Ret],
-    coro: Coroutine[object, object, T_Ret],
-    look_at_task: Callable[[asyncio.Task[T_Ret]], None] | None = None,
-) -> T_Ret:
-    __tracebackhide__ = True
-
-    def silent_done_task(res: asyncio.Future[T_Ret]) -> None:
-        if res.cancelled():
-            pass
-        res.exception()
-
-    task = loop.create_task(coro, context=ctx)
-    task.add_done_callback(silent_done_task)
-
-    if look_at_task:
-        look_at_task(task)
-
-    res = loop.run_until_complete(task)
-    _raise_maybe(func, info)
-
-    return res
-
-
-def _wrap(
-    fixturedef: pytest.FixtureDef[object],
-    extras: list[str],
-    override: Callable[..., T_Ret],
-    func: Callable[P_Args, object],
-) -> Callable[P_Args, T_Ret]:
-    """
-    Used to wrap a fixture so that we get extra information to be given to
-    override and remove those extra information from kwargs so that the override
-    function can safely execute the function being wrapped.
-
-    for example::
-
-        def my_fixture(one):
-            pass
-
-        wrapped = _wrap(fixturedef, ["request"], override, my_fixture)
-
-    can be thought of as replacing my_fixture with::
-
-        def override(extra, *args, **kwargs):
-            original_my_fixture(*args, **kwargs)
-
-        def my_fixture(one, request):
-            override({"request": ...}, one=one)
-    """
-    strip = {}
-
-    for extra in extras:
-        if extra not in fixturedef.argnames:
-            fixturedef.argnames += (extra,)  # type: ignore[misc]
-            strip[extra] = True
-
-    @wraps(func)
-    def wrapper(*args: P_Args.args, **kwargs: P_Args.kwargs) -> T_Ret:
-        __tracebackhide__ = True
-
-        got = {}
-        for extra in extras:
-            got[extra] = kwargs[extra]
-            if extra in strip:
-                del kwargs[extra]
-
-        return override(got, *args, **kwargs)
-
-    wrapper.__original__ = func  # type: ignore[attr-defined]
-
-    return wrapper
-
-
-def convert_async_coroutine_fixture(
-    ctx: contextvars.Context,
-    fixturedef: pytest.FixtureDef[object],
-    request: pytest.FixtureRequest,
-    node: pytest.Item,
-) -> None:
-    """
-    Run our async fixture in our event loop and capture the error from
-    inside the loop.
-    """
-    _func: Any = fixturedef.func
-    func: Callable[..., Awaitable[object]] = _func
-    timeout = _find_async_timeout(func, node)
-
-    def override(extra: dict[str, object], *args: object, **kwargs: object) -> object:
-        __tracebackhide__ = True
-
-        info: RunInfo = {"cancelled": False, "func": func}
-        loop = asyncio.get_event_loop_policy().get_event_loop()
-        return _run_and_raise(
-            ctx, loop, info, func, async_runner(func, timeout, info, args, kwargs)
-        )
-
-    fixturedef.func = _wrap(fixturedef, [], override, func)  # type: ignore[misc]
-
-
-def convert_async_gen_fixture(
-    ctx: contextvars.Context,
-    fixturedef: pytest.FixtureDef[object],
-    request: pytest.FixtureRequest,
-    node: pytest.Item,
-) -> None:
-    """
-    Return the yield'd value from the generator and ensure the generator is
-    finished.
-    """
-    _generator: Any = fixturedef.func
-    generator: Callable[..., AsyncGenerator[object]] = _generator
-
-    class Extra(TypedDict):
-        request: pytest.FixtureRequest
-
-    def override(extra: Extra, *args: object, **kwargs: object) -> object:
-        __tracebackhide__ = True
-
-        request = extra["request"]
-        loop = asyncio.get_event_loop_policy().get_event_loop()
-
-        timeout = _find_async_timeout(fixturedef.func, node)
-        gen_obj = generator(*args, **kwargs)
-
-        info: GenRunInfo = {"gen_obj": gen_obj, "cancelled": False}
-
-        def finalizer() -> None:
-            """Yield again, to finalize."""
-            __tracebackhide__ = True
-
-            nonlocal info
-            if "ran_once" not in info:
-                return
-
-            info = {"gen_obj": gen_obj, "finalizer": True, "cancelled": False}
-
-            async def async_finalizer() -> None:
-                __tracebackhide__ = True
-
-                await async_runner(gen_obj.__anext__, timeout, info, (), {})
-                if "e" in info:
-                    if isinstance(info["e"], StopAsyncIteration):
-                        del info["e"]
-                else:
-                    info["e"] = ValueError("Async generator fixture should only yield once")
-
-            _run_and_raise(ctx, loop, info, generator, async_finalizer())
-
-        request.addfinalizer(finalizer)
-
-        return _run_and_raise(
-            ctx,
-            loop,
-            info,
-            generator,
-            async_runner(gen_obj.__anext__, timeout, info, (), {}),
-        )
-
-    fixturedef.func = _wrap(fixturedef, ["request"], override, generator)  # type: ignore[misc]
-
-
-async def async_runner(
-    func: Callable[P_Args, Awaitable[T_Ret]],
-    timeout: float,
-    info: RunInfo,
-    args: P_Args.args,
-    kwargs: P_Args.kwargs,
-) -> T_Ret | None:
-    """
-    Run our function until timeout has been reached, at which point we cancel
-    this task. We record any exceptions in info so that calling code can extract
-    the exception from within the loop.
-    """
-    info["cancelled"] = False
-
-    current_task = asyncio.current_task()
-
-    def timeout_task(task: asyncio.Task[T_Ret] | None) -> None:
-        if task and not task.done():
-            # If the debugger is active then don't cancel, so that debugging may continue
-            # sys.gettrace is not a language feature and notguaranteed to be available
-            # on all python implementations, so we see if it exists
-            gettrace = getattr(sys, "gettrace", None)
-            if gettrace is None or gettrace() is None:
-                info["cancelled"] = True
-                task.cancel()
-
-    asyncio.get_event_loop().call_later(timeout, timeout_task, current_task)
-
-    try:
-        return await func(*args, **kwargs)
-    except:
-        __tracebackhide__ = True
-        _, e, _ = sys.exc_info()
-        if TYPE_CHECKING:
-            assert e is not None
-        info["e"] = e
-    finally:
-        info["ran_once"] = True
-
-    return None
diff --git a/alt_pytest_asyncio/base.py b/alt_pytest_asyncio/base.py
new file mode 100644
index 0000000..bd196f5
--- /dev/null
+++ b/alt_pytest_asyncio/base.py
@@ -0,0 +1,25 @@
+import abc
+from collections.abc import Callable
+from typing import NoReturn
+
+
+class AsyncTimeout(abc.ABC):
+    run_count: int
+    error: BaseException | None
+
+    @abc.abstractmethod
+    def set_timeout_seconds(self, timeout: float) -> None: ...
+
+    @abc.abstractmethod
+    def use_default_timeout(self) -> None: ...
+
+    @abc.abstractmethod
+    def raise_maybe(self, func: Callable[..., object]) -> None: ...
+
+
+class AsyncTimeoutProvider(abc.ABC):
+    @abc.abstractmethod
+    def load(self, *, default_timeout: float) -> AsyncTimeout: ...
+
+    @abc.abstractmethod
+    def set_timeout_seconds(self, timeout: float) -> NoReturn: ...
diff --git a/alt_pytest_asyncio/converter.py b/alt_pytest_asyncio/converter.py
new file mode 100644
index 0000000..715fd62
--- /dev/null
+++ b/alt_pytest_asyncio/converter.py
@@ -0,0 +1,257 @@
+import asyncio
+import contextvars
+import inspect
+import sys
+from collections import defaultdict
+from collections.abc import AsyncGenerator, Awaitable, Callable, Generator
+from functools import wraps
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+from . import base, machinery, protocols
+
+_PytestScopes = ["function", "class", "module", "package", "session"]
+
+
+class Converter:
+    def __init__(self) -> None:
+        self._ctx = contextvars.copy_context()
+        self._test_tasks: dict[asyncio.AbstractEventLoop, list[asyncio.Task[object]]] = (
+            defaultdict(list)
+        )
+
+    def sessionfinish(self) -> None:
+        for loop, tasks in self._test_tasks.items():
+            ts = []
+            for t in tasks:
+                if not t.done():
+                    t.cancel()
+                    ts.append(t)
+
+            if ts:
+                loop.run_until_complete(asyncio.tasks.gather(*ts, return_exceptions=True))
+
+    def convert_fixturedef(
+        self, fixturedef: pytest.FixtureDef[object], request: pytest.FixtureRequest
+    ) -> None:
+        if not hasattr(fixturedef, "func"):
+            return
+
+        if hasattr(fixturedef.func, "__alt_asyncio_pytest_converted__"):
+            return
+
+        if inspect.iscoroutinefunction(fixturedef.func):
+            async_timeout = self._get_async_timeout(request.scope, request.getfixturevalue)
+            self._convert_async_coroutine_fixture(fixturedef, request, async_timeout)
+
+        elif inspect.isasyncgenfunction(fixturedef.func):
+            async_timeout = self._get_async_timeout(request.scope, request.getfixturevalue)
+            self._convert_async_gen_fixture(fixturedef, request, async_timeout)
+
+        elif inspect.isgeneratorfunction(fixturedef.func):
+            self._convert_sync_gen_fixture(fixturedef)
+
+        else:
+            self._convert_sync_fixture(fixturedef)
+
+        fixturedef.func.__alt_asyncio_pytest_converted__ = True  # type: ignore[attr-defined]
+
+    def convert_pyfunc(self, pyfuncitem: pytest.Function) -> None:
+        if inspect.iscoroutinefunction(pyfuncitem.obj):
+            _obj: Any = pyfuncitem.obj
+            func: Callable[..., Awaitable[object]] = _obj
+
+            async_timeout = self._get_async_timeout(
+                "function", pyfuncitem._request.getfixturevalue
+            )
+
+            @wraps(func)
+            def run_test(*args: object, **kwargs: object) -> object:
+                res = self._run(async_timeout, func, args, kwargs)
+                async_timeout.raise_maybe(func)
+                return res
+
+            pyfuncitem.obj = run_test
+        else:
+            original: Callable[..., object] = pyfuncitem.obj
+            pyfuncitem.obj = machinery.run_sync_with_ctx(self._ctx, original)
+
+    def _convert_async_coroutine_fixture(
+        self,
+        fixturedef: pytest.FixtureDef[object],
+        request: pytest.FixtureRequest,
+        async_timeout: base.AsyncTimeout,
+    ) -> None:
+        """
+        Run our async fixture in our event loop and capture the error from
+        inside the loop.
+        """
+        _func: Any = fixturedef.func
+        func: Callable[..., Awaitable[object]] = _func
+
+        @wraps(func)
+        def run_fixture(*args: object, **kwargs: object) -> object:
+            __tracebackhide__ = True
+
+            res = self._run(async_timeout, func, args, kwargs)
+            async_timeout.raise_maybe(func)
+            return res
+
+        fixturedef.func = run_fixture  # type: ignore[misc]
+
+    def _convert_async_gen_fixture(
+        self,
+        fixturedef: pytest.FixtureDef[object],
+        request: pytest.FixtureRequest,
+        async_timeout: base.AsyncTimeout,
+    ) -> None:
+        """
+        Return the yield'd value from the generator and ensure the generator is
+        finished.
+        """
+        _generator: Any = fixturedef.func
+        generator: Callable[..., AsyncGenerator[object]] = _generator
+
+        @wraps(generator)
+        def run_fixture(*args: object, **kwargs: object) -> object:
+            __tracebackhide__ = True
+
+            if "async_timeout" in kwargs:
+                kwargs["async_timeout"] = async_timeout
+
+            gen_obj = generator(*args, **kwargs)
+
+            def finalizer() -> None:
+                """Yield again, to finalize."""
+                __tracebackhide__ = True
+
+                if not async_timeout.run_count > 0:
+                    return
+
+                async def async_finalizer() -> None:
+                    __tracebackhide__ = True
+
+                    async_timeout.use_default_timeout()
+                    await self._async_runner(async_timeout, gen_obj.__anext__, (), {})
+                    if async_timeout.error is not None:
+                        if isinstance(async_timeout.error, StopAsyncIteration):
+                            async_timeout.error = None
+                    else:
+                        async_timeout.error = ValueError(
+                            "Async generator fixture should only yield once"
+                        )
+
+                self._run(async_timeout, async_finalizer, (), {})
+                async_timeout.raise_maybe(generator)
+
+            request.addfinalizer(finalizer)
+
+            res = self._run(async_timeout, gen_obj.__anext__, (), {})
+            async_timeout.raise_maybe(generator)
+            return res
+
+        fixturedef.func = run_fixture  # type: ignore[misc]
+
+    def _convert_sync_fixture(self, fixturedef: pytest.FixtureDef[object]) -> None:
+        """
+        Used to make sure a non-async fixture is run in our
+        asyncio contextvars
+        """
+        original: Callable[..., object] = fixturedef.func
+        fixturedef.func = machinery.run_sync_with_ctx(self._ctx, original)  # type: ignore[misc]
+
+    def _convert_sync_gen_fixture(self, fixturedef: pytest.FixtureDef[object]) -> None:
+        """
+        Used to make sure a non-async generator fixture is run in our
+        asyncio contextvars
+        """
+        _original: Any = fixturedef.func
+        original: Callable[..., Generator[object, None, None]] = _original
+
+        @wraps(original)
+        def run_fixture(*args: object, **kwargs: object) -> object:
+            cm = original(*args, **kwargs)
+            value = machinery.run_sync_with_ctx(self._ctx, cm.__next__)()
+
+            try:
+                yield value
+                machinery.run_sync_with_ctx(self._ctx, cm.__next__)()
+            except StopIteration:
+                pass
+
+        fixturedef.func = run_fixture  # type: ignore[misc]
+
+    async def _async_runner(
+        self,
+        async_timeout: base.AsyncTimeout,
+        func: Callable[protocols.P_Args, Awaitable[protocols.T_Ret]],
+        args: protocols.P_Args.args,
+        kwargs: protocols.P_Args.kwargs,
+    ) -> protocols.T_Ret | None:
+        try:
+            if "async_timeout" in kwargs:
+                kwargs["async_timeout"] = async_timeout
+            async_timeout.use_default_timeout()
+            return await func(*args, **kwargs)
+        except:
+            __tracebackhide__ = True
+            _, e, _ = sys.exc_info()
+            if TYPE_CHECKING:
+                assert e is not None
+            async_timeout.error = e
+        finally:
+            async_timeout.run_count += 1
+
+        return None
+
+    def _run(
+        self,
+        async_timeout: base.AsyncTimeout,
+        func: Callable[..., Awaitable[protocols.T_Ret]],
+        args: object,
+        kwargs: object,
+    ) -> protocols.T_Ret | None:
+        __tracebackhide__ = True
+
+        def silent_done_task(res: asyncio.Future[protocols.T_Ret | None] | None) -> None:
+            if res is None:
+                return
+            if res.cancelled():
+                pass
+            res.exception()
+            return
+
+        loop = asyncio.get_event_loop_policy().get_event_loop()
+        task = loop.create_task(
+            self._async_runner(async_timeout, func, args, kwargs), context=self._ctx
+        )
+        task.add_done_callback(silent_done_task)
+        self._test_tasks[loop].append(task)
+
+        return loop.run_until_complete(task)
+
+    def _get_async_timeout(
+        self, scope: str, getfixturevalue: Callable[[str], object]
+    ) -> base.AsyncTimeout:
+        assert scope in _PytestScopes
+
+        default_timeout: float = 5
+        for scope in _PytestScopes[_PytestScopes.index(scope) :]:
+            name = "default_async_timeout"
+            if scope != "function":
+                name = f"{scope}_{name}"
+
+            try:
+                default_timeout_fix = getfixturevalue(name)
+            except pytest.FixtureLookupError:
+                pass
+            else:
+                assert isinstance(default_timeout_fix, int | float)
+                default_timeout = default_timeout_fix
+                break
+
+        async_timeout_provider = getfixturevalue("async_timeout")
+        assert isinstance(async_timeout_provider, base.AsyncTimeoutProvider)
+
+        return async_timeout_provider.load(default_timeout=default_timeout)
diff --git a/alt_pytest_asyncio/enable.py b/alt_pytest_asyncio/enable.py
new file mode 100644
index 0000000..0ff57fa
--- /dev/null
+++ b/alt_pytest_asyncio/enable.py
@@ -0,0 +1,13 @@
+import pytest
+
+from . import plugin
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    """
+    Used to configure alt_pytest_asyncio
+    """
+    if not any(
+        isinstance(p, plugin.AltPytestAsyncioPlugin) for p in config.pluginmanager.get_plugins()
+    ):
+        config.pluginmanager.register(plugin.AltPytestAsyncioPlugin())
diff --git a/alt_pytest_asyncio/errors.py b/alt_pytest_asyncio/errors.py
new file mode 100644
index 0000000..b942162
--- /dev/null
+++ b/alt_pytest_asyncio/errors.py
@@ -0,0 +1,10 @@
+class AltPytestAsyncioError(Exception):
+    pass
+
+
+class PluginAlreadyStarted(AltPytestAsyncioError):
+    pass
+
+
+class NoAsyncTimeoutInSyncFunctions(AltPytestAsyncioError):
+    pass
diff --git a/alt_pytest_asyncio/loop_manager.py b/alt_pytest_asyncio/loop_manager.py
new file mode 100644
index 0000000..d3c625f
--- /dev/null
+++ b/alt_pytest_asyncio/loop_manager.py
@@ -0,0 +1,139 @@
+import asyncio
+import contextlib
+import sys
+from collections.abc import Coroutine
+from types import TracebackType
+from typing import Self
+
+from . import machinery, protocols
+
+
+class Loop(contextlib.AbstractContextManager["Loop"]):
+    """
+    A context manager that will manager an asyncio loop and then restore the
+    original loop on exit.
+
+    Usage looks like::
+
+        from alt_pytest_asyncio import Loop, protocols
+
+        class TestThing:
+            @pytest.fixture(autouse=True)
+            def custom_loop(self) -> protocols.Loop:
+                with Loop() as custom_loop:
+                    yield custom_loop
+
+            def test_thing(self, custom_loop: protocols.Loop):
+                custom_loop.run_until_complete(my_thing())
+
+    By putting the loop into an autouse fixture, all fixtures used by the test
+    will have the custom loop. If you want to include module level fixtures too
+    then use the Loop in a module level fixture too.
+
+    Loop takes in a ``new_loop`` boolean that will make it so a new
+    loop is created. This loop will be set as `controlled_loop`.
+
+    The ``run_until_complete`` on the ``custom_loop`` in the above example will
+    do a ``run_until_complete`` on the new loop, but in a way that means you
+    won't get ``unhandled exception during shutdown`` errors when the context
+    manager closes the new loop.
+
+    When the context manager exits and closes the new loop, it will first cancel
+    all tasks to ensure finally blocks are run.
+    """
+
+    controlled_loop: asyncio.AbstractEventLoop | None
+    _original_loop: asyncio.AbstractEventLoop
+
+    def __init__(self, new_loop: bool = True) -> None:
+        self._tasks: list[asyncio.Task[object]] = []
+        self._new_loop = new_loop
+
+    def __enter__(self) -> Self:
+        self._original_loop = asyncio.get_event_loop_policy().get_event_loop()
+
+        if self._new_loop:
+            self.controlled_loop = asyncio.new_event_loop()
+        else:
+            self.controlled_loop = None
+
+        asyncio.set_event_loop(self.controlled_loop)
+        return self
+
+    def __exit__(
+        self,
+        exc_typ: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        try:
+            if self.controlled_loop is not None:
+                machinery.cancel_all_tasks(
+                    self.controlled_loop, ignore_errors_from_tasks=self._tasks
+                )
+                self.controlled_loop.run_until_complete(self.shutdown_asyncgens())
+                self.controlled_loop.close()
+        finally:
+            if original_loop := getattr(self, "_original_loop", None):
+                asyncio.set_event_loop(original_loop)
+
+    async def shutdown_asyncgens(self) -> None:
+        """
+        A version of loop.shutdown_asyncgens that tries to cancel the generators
+        before closing them.
+        """
+        if self.controlled_loop is None:
+            return
+
+        asyncgens = getattr(self.controlled_loop, "_asyncgens", None)
+        assert asyncgens is not None
+        if not len(asyncgens):
+            return
+
+        closing_agens = list(asyncgens)
+        asyncgens.clear()
+
+        # I would do an asyncio.tasks.gather but it would appear that just causes
+        # the asyncio loop to think it's shutdown, so I have to do them one at a time
+        for ag in closing_agens:
+            try:
+                try:
+                    try:
+                        await ag.athrow(asyncio.CancelledError())
+                    except StopAsyncIteration:
+                        pass
+                finally:
+                    await ag.aclose()
+            except asyncio.CancelledError:
+                pass
+            except:
+                exc = sys.exc_info()[1]
+                self.controlled_loop.call_exception_handler(
+                    {
+                        "message": "an error occurred during closing of asynchronous generator",
+                        "exception": exc,
+                        "asyncgen": ag,
+                    }
+                )
+
+    def run_until_complete(
+        self, coro: Coroutine[object, object, protocols.T_Ret]
+    ) -> protocols.T_Ret:
+        if not hasattr(self, "controlled_loop"):
+            raise Exception(
+                "Cannot use run_until_complete on this alt_pytest_asyncio.Loop outside of using it as a context manager"
+            )
+
+        if self.controlled_loop is None:
+            raise Exception(
+                "This alt_pytest_asyncio.Loop is not managing your overridden loop, use run_until_complete on that loop instead"
+            )
+
+        task = self.controlled_loop.create_task(coro)
+
+        # Add the task so that when we cancel all tasks before closing the loop
+        # We don't complain about errors in this particular task
+        # As we get the errors risen to the caller via run_until_complete
+        self._tasks.append(task)
+
+        return self.controlled_loop.run_until_complete(task)
diff --git a/alt_pytest_asyncio/machinery.py b/alt_pytest_asyncio/machinery.py
new file mode 100644
index 0000000..bb3c25f
--- /dev/null
+++ b/alt_pytest_asyncio/machinery.py
@@ -0,0 +1,108 @@
+import asyncio
+import contextvars
+import dataclasses
+import functools
+import sys
+from collections.abc import Callable, Coroutine
+from typing import TYPE_CHECKING, ParamSpec, TypeVar
+
+from _pytest._code.code import ExceptionInfo
+
+P_Args = ParamSpec("P_Args")
+T_Ret = TypeVar("T_Ret")
+
+
+def cancel_all_tasks(
+    loop: asyncio.AbstractEventLoop,
+    ignore_errors_from_tasks: list[asyncio.Task[object]] | None = None,
+) -> None:
+    to_cancel = asyncio.tasks.all_tasks(loop)
+
+    if not to_cancel:
+        return
+
+    for task in to_cancel:
+        task.cancel()
+
+    gathered = asyncio.tasks.gather(*to_cancel, return_exceptions=True)
+    loop.run_until_complete(gathered)
+
+    for task in to_cancel:
+        if task.cancelled():
+            continue
+
+        if task.exception() is not None:
+            if ignore_errors_from_tasks and task in ignore_errors_from_tasks:
+                continue
+
+            loop.call_exception_handler(
+                {
+                    "message": "unhandled exception during shutdown",
+                    "exception": task.exception(),
+                    "task": task,
+                }
+            )
+
+
+def run_coro_as_main(
+    loop: asyncio.AbstractEventLoop, coro: Coroutine[object, object, None]
+) -> None:
+    @dataclasses.dataclass(frozen=True, kw_only=True)
+    class Captured(Exception):
+        error: BaseException
+
+    try:
+
+        async def runner() -> None:
+            __tracebackhide__ = True
+
+            try:
+                await coro
+            except:
+                exc_type, exc, tb = sys.exc_info()
+                if TYPE_CHECKING:
+                    assert exc_type is not None
+                    assert exc is not None
+                    assert tb is not None
+
+                exc.__traceback__ = tb
+                raise Captured(error=exc)
+
+        task = loop.create_task(runner())
+        loop.run_until_complete(task)
+    except:
+        exc_type, exc, tb = sys.exc_info()
+        if (
+            isinstance(exc_type, type)
+            and issubclass(exc_type, Captured)
+            and isinstance(exc, Captured)
+        ):
+            exc = exc.error
+            exc_type = type(exc)
+            tb = exc.__traceback__
+
+        if TYPE_CHECKING:
+            assert exc_type is not None
+            assert exc is not None
+            assert tb is not None
+
+        info = ExceptionInfo[BaseException]((exc_type, exc, tb), "")
+        sys.exit(str(info.getrepr(style="short")))
+    finally:
+        cancel_all_tasks(loop, ignore_errors_from_tasks=[task])
+        loop.close()
+
+
+def run_sync_with_ctx(
+    ctx: contextvars.Context, func: Callable[P_Args, T_Ret]
+) -> Callable[P_Args, T_Ret]:
+    @functools.wraps(func)
+    def run(*args: P_Args.args, **kwargs: P_Args.kwargs) -> T_Ret:
+        try:
+            ctx.run(lambda: None)
+        except RuntimeError:
+            return func(*args, **kwargs)
+        else:
+            return ctx.run(func, *args, **kwargs)
+
+    return run
diff --git a/alt_pytest_asyncio/plugin.py b/alt_pytest_asyncio/plugin.py
index da742e8..50fb24b 100644
--- a/alt_pytest_asyncio/plugin.py
+++ b/alt_pytest_asyncio/plugin.py
@@ -1,318 +1,204 @@
 import asyncio
-import contextvars
-import dataclasses
+import contextlib
 import inspect
 import sys
-from collections import defaultdict
-from collections.abc import Callable, Coroutine, Iterator
-from functools import partial, wraps
+from collections.abc import Callable, Iterator
 from types import TracebackType
-from typing import TYPE_CHECKING, Self
+from typing import TYPE_CHECKING, NoReturn, cast
 
 import pytest
-from _pytest._code.code import ExceptionInfo
 
-from alt_pytest_asyncio.async_converters import convert_fixtures, converted_async_test
+from . import base, converter, errors, loop_manager, protocols
 
 
-class AltPytestAsyncioPlugin:
-    def __init__(self, loop: asyncio.AbstractEventLoop | None = None) -> None:
-        self.own_loop = False
-        self.test_tasks: dict[asyncio.AbstractEventLoop, list[asyncio.Task[object]]] = defaultdict(
-            list
-        )
+@pytest.hookimpl
+def pytest_addoption(parser: pytest.Parser) -> None:
+    """
+    Will add a pytest option for default async timeouts
+    """
+    desc = "timeout in seconds before failing a test. This can be overridden with the ``async_timeout`` fixture"
 
-        if loop is None:
-            self.own_loop = True
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
+    group = parser.getgroup(
+        "alt-pytest-asyncio", description="Alternative asyncio pytest plugin options"
+    )
+    group.addoption("--default-async-timeout", type=float, dest="default_async_timeout", help=desc)
+    parser.addini("default_async_timeout", desc, default=5)
 
+
+class _ManagedLoop(contextlib.AbstractContextManager[None]):
+    _original_loop: asyncio.AbstractEventLoop | None
+
+    def __init__(self, *, loop: asyncio.AbstractEventLoop) -> None:
         self.loop = loop
-        self.ctx = contextvars.copy_context()
 
-    def pytest_configure(self, config: pytest.Config) -> None:
-        """Register our timeout marker which is used to signify async timeouts"""
-        config.addinivalue_line(
-            "markers", "async_timeout(length): mark async test to have a timeout"
-        )
+    def __enter__(self) -> None:
+        if hasattr(self, "_original_loop"):
+            raise Exception("This context manager is already active")
+
+        try:
+            self._original_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            self._original_loop = None
+
+        asyncio.set_event_loop(self.loop)
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        if not hasattr(self, "_original_loop"):
+            return
+
+        if self._original_loop is None:
+            asyncio.set_event_loop(None)
+        else:
+            asyncio.set_event_loop(self._original_loop)
+
+        del self._original_loop
+
+
+class AltPytestAsyncioPlugin:
+    def __init__(self, *, managed_loop: asyncio.AbstractEventLoop | None = None) -> None:
+        self._managed_loop = managed_loop
+        self._converter = converter.Converter()
+
+    @pytest.hookimpl(tryfirst=True, hookwrapper=True)
+    def pytest_sessionstart(self, session: pytest.Session) -> Iterator[None]:
+        if hasattr(self, "_cm"):
+            raise errors.PluginAlreadyStarted()
+
+        self._cm = contextlib.ExitStack()
+        if self._managed_loop is None:
+            self._cm.enter_context(loop_manager.Loop(new_loop=True))
+        else:
+            self._cm.enter_context(_ManagedLoop(loop=self._managed_loop))
+
+        yield
 
     @pytest.hookimpl(tryfirst=True, hookwrapper=True)
     def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int) -> Iterator[None]:
         """
         Make sure all the test coroutines have been finalized once pytest has finished
 
-        Also, if we created our own loop, then cancel any remaining tasks on it and close it.
-
         This is so if pytest is interrupted, we still execute the finally blocks of all the tests
         """
         try:
-            for loop, tasks in self.test_tasks.items():
-                ts = []
-                for t in tasks:
-                    if not t.done():
-                        t.cancel()
-                        ts.append(t)
-
-                if ts:
-                    self.loop.run_until_complete(asyncio.tasks.gather(*ts, return_exceptions=True))
+            self._converter.sessionfinish()
             yield
         finally:
-            if self.own_loop:
-                try:
-                    cancel_all_tasks(self.loop)
-                finally:
-                    self.loop.close()
+            if _cm := getattr(self, "_cm", None):
+                _cm.close()
 
     @pytest.hookimpl(tryfirst=True, hookwrapper=True)
     def pytest_fixture_setup(
         self, fixturedef: pytest.FixtureDef[object], request: pytest.FixtureRequest
     ) -> Iterator[None]:
         """Convert async fixtures to sync fixtures"""
-        convert_fixtures(self.ctx, fixturedef, request, request.node)
+        self._converter.convert_fixturedef(fixturedef, request)
         yield
 
     @pytest.hookimpl(tryfirst=True, hookwrapper=True)
     def pytest_pyfunc_call(self, pyfuncitem: pytest.Function) -> Iterator[None]:
         """Convert async tests to sync tests"""
-        if inspect.iscoroutinefunction(pyfuncitem.obj):
-            timeout_marker = pyfuncitem.get_closest_marker("async_timeout")
-
-            if timeout_marker:
-                timeout = float(timeout_marker.args[0])
-            else:
-                timeout = float(
-                    pyfuncitem.config.getoption("default_async_timeout", None)
-                    or pyfuncitem.config.getini("default_async_timeout")
-                )
-
-            o = pyfuncitem.obj
-            pyfuncitem.obj = wraps(o)(
-                partial(converted_async_test, self.ctx, self.test_tasks, o, timeout)
-            )
-        else:
-            original: Callable[..., object] = pyfuncitem.obj
-
-            @wraps(original)
-            def run_obj(*args: object, **kwargs: object) -> None:
-                try:
-                    self.ctx.run(lambda: None)
-
-                    def run(func: Callable[..., object]) -> object:
-                        return self.ctx.run(func)
-
-                except RuntimeError:
-
-                    def run(func: Callable[..., object]) -> object:
-                        return func()
-
-                run(partial(original, *args, **kwargs))
-
-            pyfuncitem.obj = run_obj
-
+        self._converter.convert_pyfunc(pyfuncitem)
         yield
 
 
-def cancel_all_tasks(
-    loop: asyncio.AbstractEventLoop,
-    ignore_errors_from_tasks: list[asyncio.Task[object]] | None = None,
-) -> None:
-    to_cancel = asyncio.tasks.all_tasks(loop)
-
-    if not to_cancel:
-        return
-
-    for task in to_cancel:
-        task.cancel()
-
-    gathered = asyncio.tasks.gather(*to_cancel, return_exceptions=True)
-    loop.run_until_complete(gathered)
-
-    for task in to_cancel:
-        if task.cancelled():
-            continue
-
-        if task.exception() is not None:
-            if ignore_errors_from_tasks and task in ignore_errors_from_tasks:
-                continue
-
-            loop.call_exception_handler(
-                {
-                    "message": "unhandled exception during shutdown",
-                    "exception": task.exception(),
-                    "task": task,
-                }
-            )
-
-
-def run_coro_as_main(
-    loop: asyncio.AbstractEventLoop, coro: Coroutine[object, object, None]
-) -> None:
-    @dataclasses.dataclass(frozen=True, kw_only=True)
-    class Captured(Exception):
-        error: BaseException
-
-    try:
-
-        async def runner() -> None:
-            __tracebackhide__ = True
-
-            try:
-                await coro
-            except:
-                exc_type, exc, tb = sys.exc_info()
-                if TYPE_CHECKING:
-                    assert exc_type is not None
-                    assert exc is not None
-                    assert tb is not None
-
-                exc.__traceback__ = tb
-                raise Captured(error=exc)
-
-        task = loop.create_task(runner())
-        loop.run_until_complete(task)
-    except:
-        exc_type, exc, tb = sys.exc_info()
-        if (
-            isinstance(exc_type, type)
-            and issubclass(exc_type, Captured)
-            and isinstance(exc, Captured)
-        ):
-            exc = exc.error
-            exc_type = type(exc)
-            tb = exc.__traceback__
-
-        if TYPE_CHECKING:
-            assert exc_type is not None
-            assert exc is not None
-            assert tb is not None
-
-        info = ExceptionInfo[BaseException]((exc_type, exc, tb), "")
-        print(info.getrepr(style="short"))
-        sys.exit(1)
-    finally:
-        cancel_all_tasks(loop, ignore_errors_from_tasks=[task])
-        loop.close()
-
-
-class OverrideLoop:
-    """
-    A context manager that will install a new asyncio loop and then restore the
-    original loop on exit.
-
-    Usage looks like::
+class LoadedAsyncTimeout(base.AsyncTimeout):
+    def __init__(self, *, default_timeout: float) -> None:
+        self.error: BaseException | None = None
+        self.timeout: float = default_timeout
+        self.cancelled: bool = False
+        self.run_count: int = 0
+        self._timeout: asyncio.TimerHandle | None = None
 
-        from alt_pytest_asyncio.plugin import OverrideLoop
+    def use_default_timeout(self) -> None:
+        self.set_timeout_seconds(self.timeout)
 
-        class TestThing:
-            @pytest.fixture(autouse=True)
-            def custom_loop(self):
-                with OverrideLoop() as custom_loop:
-                    yield custom_loop
+    def set_timeout_seconds(self, timeout: float) -> None:
+        if self._timeout:
+            self._timeout.cancel()
 
-            def test_thing(self, custom_loop):
-                custom_loop.run_until_complete(my_thing())
+        self.timeout = timeout
+        current_task = asyncio.current_task()
 
-    By putting the loop into an autouse fixture, all fixtures used by the test
-    will have the custom loop. If you want to include module level fixtures too
-    then use the OverrideLoop in a module level fixture too.
+        def timeout_task(task: asyncio.Task[object] | None) -> None:
+            if timeout < self.timeout:
+                return
 
-    OverrideLoop takes in a ``new_loop`` boolean that will make it so no new
-    loop is set and asyncio is left with no default loop.
+            if task and not task.done():
+                # If the debugger is active then don't cancel, so that debugging may continue
+                # sys.gettrace is not a language feature and not guaranteed to be available
+                # on all python implementations, so we see if it exists
+                gettrace = getattr(sys, "gettrace", None)
+                if gettrace is None or gettrace() is None:
+                    self.cancelled = True
+                    task.cancel()
 
-    The new loop itself (or None if new_loop is False) can be found in the
-    ``loop`` attribute of the object yielded by the context manager.
+        self._timeout = asyncio.get_event_loop().call_later(timeout, timeout_task, current_task)
 
-    The ``run_until_complete`` on the ``custom_loop`` in the above example will
-    do a ``run_until_complete`` on the new loop, but in a way that means you
-    won't get ``unhandled exception during shutdown`` errors when the context
-    manager closes the new loop.
+    def raise_maybe(self, func: Callable[..., object]) -> None:
+        __tracebackhide__ = True
 
-    When the context manager exits and closes the new loop, it will first cancel
-    all tasks to ensure finally blocks are run.
-    """
+        if hasattr(func, "__original__"):
+            func = func.__original__
 
-    loop: asyncio.AbstractEventLoop | None
+        fle = inspect.getfile(func)
+        if hasattr(func, "__func__"):
+            func = func.__func__
+        lineno = func.__code__.co_firstlineno
 
-    def __init__(self, new_loop: bool = True) -> None:
-        self.tasks: list[asyncio.Task[object]] = []
-        self.new_loop = new_loop
+        def raise_error() -> None:
+            if self.cancelled:
+                raise AssertionError(
+                    f"Took too long to complete: {fle}:{lineno} (timeout={self.timeout})"
+                )
+            if self.error:
+                raise self.error
 
-    def __enter__(self) -> Self:
-        self._original_loop = asyncio.get_event_loop_policy().get_event_loop()
+        if self.error:
+            # Use a separate function so when --tb=short is not set we don't get
+            # this entire function in the output
+            raise_error()
 
-        if self.new_loop:
-            self.loop = asyncio.new_event_loop()
-        else:
-            self.loop = None
 
-        asyncio.set_event_loop(self.loop)
-        return self
+class AsyncTimeoutProvider(base.AsyncTimeoutProvider):
+    def __init__(self, timeout_factory: protocols.AsyncTimeoutFactory) -> None:
+        self.timeout_factory = timeout_factory
 
-    def __exit__(self, exc_typ: type[Exception], exc: Exception, tb: TracebackType) -> None:
-        try:
-            if self.loop is not None:
-                cancel_all_tasks(self.loop, ignore_errors_from_tasks=self.tasks)
-                self.loop.run_until_complete(self.shutdown_asyncgens())
-                self.loop.close()
-        finally:
-            if hasattr(self, "_original_loop"):
-                asyncio.set_event_loop(self._original_loop)
+    def load(self, *, default_timeout: float) -> base.AsyncTimeout:
+        return self.timeout_factory(default_timeout=default_timeout)
 
-    async def shutdown_asyncgens(self) -> None:
-        """
-        A version of loop.shutdown_asyncgens that tries to cancel the generators
-        before closing them.
-        """
-        if self.loop is None:
-            return
+    def set_timeout_seconds(self, timeout: float) -> NoReturn:
+        raise errors.NoAsyncTimeoutInSyncFunctions(
+            "The async_timeout fixture only makes sense in async fixtures/functions"
+        )
 
-        asyncgens = getattr(self.loop, "_asyncgens", None)
-        assert asyncgens is not None
-        if not len(asyncgens):
-            return
 
-        closing_agens = list(asyncgens)
-        asyncgens.clear()
-
-        # I would do an asyncio.tasks.gather but it would appear that just causes
-        # the asyncio loop to think it's shutdown, so I have to do them one at a time
-        for ag in closing_agens:
-            try:
-                try:
-                    try:
-                        await ag.athrow(asyncio.CancelledError())
-                    except StopAsyncIteration:
-                        pass
-                finally:
-                    await ag.aclose()
-            except asyncio.CancelledError:
-                pass
-            except:
-                exc = sys.exc_info()[1]
-                self.loop.call_exception_handler(
-                    {
-                        "message": "an error occurred during closing of asynchronous generator",
-                        "exception": exc,
-                        "asyncgen": ag,
-                    }
-                )
+@pytest.fixture(scope="session")
+def session_default_async_timeout(pytestconfig: pytest.Config) -> float:
+    timeout = pytestconfig.option.default_async_timeout
+    if timeout is None:
+        timeout = 5
+    return float(timeout)
 
-    def run_until_complete(self, coro: Coroutine[object, object, None]) -> None:
-        if not hasattr(self, "loop"):
-            raise Exception(
-                "Cannot use run_until_complete on OverrideLoop outside of using it as a context manager"
-            )
 
-        if self.loop is None:
-            raise Exception(
-                "OverrideLoop is not managing your overridden loop, use run_until_complete on that loop instead"
-            )
+@pytest.fixture(scope="session")
+def async_timeout() -> protocols.AsyncTimeoutProvider:
+    """
+    This is a special fixture where asking for it in a fixture or a test provides
+    an object that matches ``alt_pytest_asyncio.protocols.AsyncTimeout``
 
-        task = self.loop.create_task(coro)
+    This can be overridden to return an object that sets a different ``load``
+    method if that's desirable.
+    """
+    return AsyncTimeoutProvider(timeout_factory=LoadedAsyncTimeout)
 
-        # Add the task so that when we cancel all tasks before closing the loop
-        # We don't complain about errors in this particular task
-        # As we get the errors risen to the caller via run_until_complete
-        self.tasks.append(task)
 
-        return self.loop.run_until_complete(task)
+if TYPE_CHECKING:
+    _ATP: protocols.AsyncTimeoutProvider = cast(AsyncTimeoutProvider, None)
+    _ATPF: protocols.AsyncTimeout = cast(AsyncTimeoutProvider, None)
+    _AT: protocols.AsyncTimeout = cast(LoadedAsyncTimeout, None)
+    _ATF: protocols.AsyncTimeoutFactory = LoadedAsyncTimeout
diff --git a/alt_pytest_asyncio/protocols.py b/alt_pytest_asyncio/protocols.py
new file mode 100644
index 0000000..c19261b
--- /dev/null
+++ b/alt_pytest_asyncio/protocols.py
@@ -0,0 +1,31 @@
+import asyncio
+from collections.abc import Coroutine
+from typing import TYPE_CHECKING, NoReturn, ParamSpec, Protocol, TypeVar, cast
+
+from . import base
+
+T_Ret = TypeVar("T_Ret")
+P_Args = ParamSpec("P_Args")
+
+
+class Loop(Protocol):
+    controlled_loop: asyncio.AbstractEventLoop | None
+
+    def run_until_complete(self, coro: Coroutine[object, object, T_Ret]) -> T_Ret: ...
+
+
+class AsyncTimeout(Protocol):
+    def set_timeout_seconds(self, timeout: float) -> None: ...
+
+
+class AsyncTimeoutFactory(Protocol):
+    def __call__(self, *, default_timeout: float) -> base.AsyncTimeout: ...
+
+
+class AsyncTimeoutProvider(Protocol):
+    def load(self, *, default_timeout: float) -> base.AsyncTimeout: ...
+    def set_timeout_seconds(self, timeout: float) -> NoReturn: ...
+
+
+if TYPE_CHECKING:
+    _ATP: AsyncTimeout = cast(AsyncTimeoutProvider, None)
diff --git a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/conftest.py b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/conftest.py
index 2128cd0..6d9fd98 100644
--- a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/conftest.py
+++ b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/conftest.py
@@ -3,24 +3,28 @@
 
 import pytest
 
+import alt_pytest_asyncio
+
+AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
+
 
 @pytest.fixture(scope="session")
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout_in_setup_session() -> AsyncGenerator[int]:
+async def fixture_timeout_in_setup_session(async_timeout: AsyncTimeout) -> AsyncGenerator[int]:
+    async_timeout.set_timeout_seconds(0.01)
     await asyncio.sleep(1)
     yield 1
 
 
 @pytest.fixture(scope="session")
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout_session() -> int:
+async def fixture_timeout_session(async_timeout: AsyncTimeout) -> int:
+    async_timeout.set_timeout_seconds(0.01)
     await asyncio.sleep(1)
     return 1
 
 
 @pytest.fixture(scope="session", autouse=True)
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout_in_teardown_session() -> AsyncGenerator[int]:
+async def fixture_timeout_in_teardown_session(async_timeout: AsyncTimeout) -> AsyncGenerator[int]:
+    async_timeout.set_timeout_seconds(0.01)
     try:
         yield 1
     finally:
diff --git a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/expected b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/expected
index b76a6ed..9c3f67a 100644
--- a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/expected
+++ b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/expected
@@ -1,92 +1,72 @@
 =======* ERRORS =====*
 _______* ERROR at teardown of test_one ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:7
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:11 (timeout=0.01)
 _______* ERROR at setup of test_two ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:16
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:21 (timeout=0.01)
 _______* ERROR at setup of test_three ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:23
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:28 (timeout=0.01)
 _______* ERROR at setup of test_four ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:30
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:35 (timeout=0.01)
 _______* ERROR at setup of test_five ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:46
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails.py:51 (timeout=0.01)
 _______* ERROR at setup of test_six ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/conftest.py:7
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/conftest.py:11 (timeout=0.01)
 _______* ERROR at setup of test_seven ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/conftest.py:14
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/conftest.py:18 (timeout=0.01)
 _______* ERROR at teardown of test_seven ____*
-*/async_converters.py:* in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */test_fails.py:37
+*/plugin.py:* in raise_error
+E   AssertionError: Took too long to complete: */test_fails.py:42 (timeout=0.01)
 _______* ERROR at setup of TestAClass.test_2one ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_method_fixtures.py:8
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_method_fixtures.py:12 (timeout=0.01)
 _______* ERROR at setup of TestAClass.test_2two ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_method_fixtures.py:14
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_method_fixtures.py:18 (timeout=0.01)
 _______* ERROR at teardown of TestAClass.test_2three ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_method_fixtures.py:20
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_method_fixtures.py:24 (timeout=0.01)
 _______* ERROR at teardown of test_one ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:9
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:17 (timeout=0.01)
 _______* ERROR at setup of test_two ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:17
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:25 (timeout=0.01)
 _______* ERROR at setup of test_three ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:23
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:31 (timeout=0.01)
 _______* ERROR at setup of test_four ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:29
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:37 (timeout=0.02)
 _______* ERROR at setup of test_five ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:43
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_surrounding_pytestmark.py:51 (timeout=0.02)
 _______* ERROR at setup of test_six ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/conftest.py:7
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/conftest.py:11 (timeout=0.01)
 _______* ERROR at setup of test_seven ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/conftest.py:14
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/conftest.py:18 (timeout=0.01)
 _______* ERROR at teardown of test_seven ____*
   + Exception Group Traceback (most recent call last):
   |*ExceptionGroup: errors during test teardown (2 sub-exceptions)
   +-+---------------- 1 ----------------
     | Traceback (most recent call last):
-    | AssertionError: Took too long to complete: */conftest.py:21
+    | AssertionError: Took too long to complete: */conftest.py:25 (timeout=0.01)
     +---------------- 2 ----------------
     | Traceback (most recent call last):
-    | AssertionError: Took too long to complete: */test_surrounding_pytestmark.py:35
+    | AssertionError: Took too long to complete: */test_surrounding_pytestmark.py:43 (timeout=0.02)
     +------------------------------------
 =======* FAILURES ====*
 _______* test_takes_closest_pytestmark ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_tests.py:8
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_tests.py:15 (timeout=0.01)
 _______* test_takes_pytestmark_on_function2 ____*
-*/async_converters.py:*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_tests.py:17
+*/plugin.py:*: in raise_error
+E   AssertionError: Took too long to complete: */example_timeouts*/test_fails_tests.py:24 (timeout=0.02)
diff --git a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails.py b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails.py
index 6d80194..d7a3a2f 100644
--- a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails.py
+++ b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails.py
@@ -3,10 +3,15 @@
 
 import pytest
 
+import alt_pytest_asyncio
+
+AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
+
 
 @pytest.fixture()
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout_in_finally() -> AsyncGenerator[int]:
+async def fixture_timeout_in_finally(async_timeout: AsyncTimeout) -> AsyncGenerator[int]:
+    async_timeout.set_timeout_seconds(0.01)
+
     try:
         yield 1
     finally:
@@ -14,29 +19,29 @@ async def fixture_timeout_in_finally() -> AsyncGenerator[int]:
 
 
 @pytest.fixture()
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout_in_setup() -> AsyncGenerator[int]:
+async def fixture_timeout_in_setup(async_timeout: AsyncTimeout) -> AsyncGenerator[int]:
+    async_timeout.set_timeout_seconds(0.01)
     await asyncio.sleep(1)
     yield 1
 
 
 @pytest.fixture()
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout() -> int:
+async def fixture_timeout(async_timeout: AsyncTimeout) -> int:
+    async_timeout.set_timeout_seconds(0.01)
     await asyncio.sleep(1)
     return 1
 
 
 @pytest.fixture(scope="module")
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout_in_setup_module() -> AsyncGenerator[int]:
+async def fixture_timeout_in_setup_module(async_timeout: AsyncTimeout) -> AsyncGenerator[int]:
+    async_timeout.set_timeout_seconds(0.01)
     await asyncio.sleep(1)
     yield 1
 
 
 @pytest.fixture(scope="module", autouse=True)
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout_in_teardown_module() -> AsyncGenerator[int]:
+async def fixture_timeout_in_teardown_module(async_timeout: AsyncTimeout) -> AsyncGenerator[int]:
+    async_timeout.set_timeout_seconds(0.01)
     try:
         yield 1
     finally:
@@ -44,8 +49,8 @@ async def fixture_timeout_in_teardown_module() -> AsyncGenerator[int]:
 
 
 @pytest.fixture(scope="module")
-@pytest.mark.async_timeout(0.01)
-async def fixture_timeout_module() -> int:
+async def fixture_timeout_module(async_timeout: AsyncTimeout) -> int:
+    async_timeout.set_timeout_seconds(0.01)
     await asyncio.sleep(1)
     return 1
 
diff --git a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails_method_fixtures.py b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails_method_fixtures.py
index d87f5b4..8806e58 100644
--- a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails_method_fixtures.py
+++ b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails_method_fixtures.py
@@ -3,23 +3,29 @@
 
 import pytest
 
+import alt_pytest_asyncio
+
+AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
+
 
 class TestAClass:
     @pytest.fixture()
-    @pytest.mark.async_timeout(0.01)
-    async def fixture_timeout(self) -> int:
+    async def fixture_timeout(self, async_timeout: AsyncTimeout) -> int:
+        async_timeout.set_timeout_seconds(0.01)
         await asyncio.sleep(1)
         return 1
 
     @pytest.fixture()
-    @pytest.mark.async_timeout(0.01)
-    async def fixture_timeout_in_setup(self) -> AsyncGenerator[int]:
+    async def fixture_timeout_in_setup(self, async_timeout: AsyncTimeout) -> AsyncGenerator[int]:
+        async_timeout.set_timeout_seconds(0.01)
         await asyncio.sleep(1)
         yield 1
 
     @pytest.fixture()
-    @pytest.mark.async_timeout(0.01)
-    async def fixture_timeout_in_teardown(self) -> AsyncGenerator[int]:
+    async def fixture_timeout_in_teardown(
+        self, async_timeout: AsyncTimeout
+    ) -> AsyncGenerator[int]:
+        async_timeout.set_timeout_seconds(0.01)
         try:
             yield 1
         finally:
diff --git a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails_tests.py b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails_tests.py
index a8ca68e..529abb5 100644
--- a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails_tests.py
+++ b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_fails_tests.py
@@ -2,18 +2,25 @@
 
 import pytest
 
-pytestmark = pytest.mark.async_timeout(0.01)
+import alt_pytest_asyncio
+
+AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
+
+
+@pytest.fixture
+def default_async_timeout() -> float:
+    return 0.01
 
 
 async def test_takes_closest_pytestmark() -> None:
     await asyncio.sleep(0.03)
 
 
-@pytest.mark.async_timeout(0.04)
-async def test_takes_pytestmark_on_function() -> None:
+async def test_takes_pytestmark_on_function(async_timeout: AsyncTimeout) -> None:
+    async_timeout.set_timeout_seconds(0.04)
     await asyncio.sleep(0.03)
 
 
-@pytest.mark.async_timeout(0.02)
-async def test_takes_pytestmark_on_function2() -> None:
+async def test_takes_pytestmark_on_function2(async_timeout: AsyncTimeout) -> None:
+    async_timeout.set_timeout_seconds(0.02)
     await asyncio.sleep(0.03)
diff --git a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_surrounding_pytestmark.py b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_surrounding_pytestmark.py
index 1f0a8af..a3c0d02 100644
--- a/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_surrounding_pytestmark.py
+++ b/helpers/alt_pytest_asyncio_test_driver/examples/example_timeouts/test_surrounding_pytestmark.py
@@ -3,7 +3,15 @@
 
 import pytest
 
-pytestmark = pytest.mark.async_timeout(0.01)
+
+@pytest.fixture(scope="module")
+def module_default_async_timeout() -> float:
+    return 0.02
+
+
+@pytest.fixture
+def default_async_timeout() -> float:
+    return 0.01
 
 
 @pytest.fixture()
diff --git a/helpers/alt_pytest_asyncio_test_driver/examples/interrupt_test/expected b/helpers/alt_pytest_asyncio_test_driver/examples/interrupt_test/expected
index 3ae2dfa..9b0f7ab 100644
--- a/helpers/alt_pytest_asyncio_test_driver/examples/interrupt_test/expected
+++ b/helpers/alt_pytest_asyncio_test_driver/examples/interrupt_test/expected
@@ -9,9 +9,8 @@ _______* test_bb_shows_failed_tests ____*
 *tests/test_example.py*: in test_bb_shows_failed_tests
 E   AssertionError: NOOOOO
 _______* test_cc_shows_timedout_tests ____*
-*/async_converters.py*: in raise_error
-    assert False, f"Took too long to complete: {fle}:{lineno}"
-E   AssertionError: Took too long to complete: */test_example.py:*
+*/plugin.py*: in raise_error
+E   AssertionError: Took too long to complete: */test_example.py:* (timeout=1)
 =======* short test summary info ====*
 FAILED tests/test_example.py::test_bb_shows_failed_tests - AssertionError: NO*
 FAILED tests/test_example.py::test_cc_shows_timedout_tests - AssertionError: *
diff --git a/helpers/alt_pytest_asyncio_test_driver/examples/interrupt_test/tests/test_example.py b/helpers/alt_pytest_asyncio_test_driver/examples/interrupt_test/tests/test_example.py
index 2277c4b..7255911 100644
--- a/helpers/alt_pytest_asyncio_test_driver/examples/interrupt_test/tests/test_example.py
+++ b/helpers/alt_pytest_asyncio_test_driver/examples/interrupt_test/tests/test_example.py
@@ -3,6 +3,10 @@
 
 import pytest
 
+import alt_pytest_asyncio
+
+AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
+
 
 @pytest.fixture()
 async def thing() -> AsyncGenerator[int]:
@@ -20,8 +24,8 @@ async def test_bb_shows_failed_tests() -> None:
     assert False, "NOOOOO"
 
 
-@pytest.mark.async_timeout(1)
-async def test_cc_shows_timedout_tests() -> None:
+async def test_cc_shows_timedout_tests(async_timeout: AsyncTimeout) -> None:
+    async_timeout.set_timeout_seconds(1)
     await asyncio.sleep(3)
 
 
diff --git a/tests/conftest.py b/tests/conftest.py
index c6481d5..e5ef7e2 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,1 +1,1 @@
-pytest_plugins = ["pytester"]
+pytest_plugins = ["pytester", "alt_pytest_asyncio.enable"]
diff --git a/tests/test_examples.py b/tests/test_examples.py
index 10f2f1d..08c65b1 100644
--- a/tests/test_examples.py
+++ b/tests/test_examples.py
@@ -15,6 +15,10 @@
 import pytest
 from alt_pytest_asyncio_test_driver import available_examples
 
+import alt_pytest_asyncio
+
+AsyncTimeout = alt_pytest_asyncio.protocols.AsyncTimeout
+
 
 @contextlib.contextmanager
 def listening() -> Iterator[tuple[socket.socket, str]]:
@@ -77,7 +81,7 @@ async def test_shows_correctly_for_failing_fixtures(name: str, pytester: pytest.
     with importlib.resources.as_file(examples) as examples_path:
         shutil.copytree(examples_path, pytester.path / name)
 
-    result = pytester.runpytest_subprocess("--tb", "short")
+    result = pytester.runpytest_subprocess("--tb", "short", "-p", "alt_pytest_asyncio.enable")
     assert not result.errlines
 
     lines: int | list[str] = 0
@@ -97,9 +101,12 @@ async def test_shows_correctly_for_failing_fixtures(name: str, pytester: pytest.
     matcher.fnmatch_lines(expected.strip().split("\n"))
 
 
-@pytest.mark.async_timeout(7)
 @pytest.mark.skipif(os.name == "nt", reason="Can't use async subprocess on windows")
-async def test_cleans_up_tests_properly_on_interrupt(pytester: pytest.Pytester) -> None:
+async def test_cleans_up_tests_properly_on_interrupt(
+    async_timeout: AsyncTimeout, pytester: pytest.Pytester
+) -> None:
+    async_timeout.set_timeout_seconds(7)
+
     examples = (
         importlib.resources.files("alt_pytest_asyncio_test_driver") / "examples" / "interrupt_test"
     )
@@ -111,6 +118,8 @@ async def test_cleans_up_tests_properly_on_interrupt(pytester: pytest.Pytester)
             sys.executable,
             "-m",
             "pytest",
+            "-p",
+            "alt_pytest_asyncio.enable",
             cwd=directory,
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
diff --git a/tests/test_override_loop.py b/tests/test_loop_manager.py
similarity index 90%
rename from tests/test_override_loop.py
rename to tests/test_loop_manager.py
index e597b69..17ba284 100644
--- a/tests/test_override_loop.py
+++ b/tests/test_loop_manager.py
@@ -3,7 +3,7 @@
 
 import pytest
 
-from alt_pytest_asyncio.plugin import OverrideLoop
+from alt_pytest_asyncio import Loop
 
 
 def get_event_loop() -> asyncio.AbstractEventLoop:
@@ -69,10 +69,10 @@ async def blah() -> None:
 
     try:
         coro = info["coro"] = blah()
-        OverrideLoop().run_until_complete(coro)
+        Loop().run_until_complete(coro)
         assert False, "should have risen an error"
     except Exception as error:
-        msg = "Cannot use run_until_complete on OverrideLoop outside of using it as a context manager"
+        msg = "Cannot use run_until_complete on this alt_pytest_asyncio.Loop outside of using it as a context manager"
         assert str(error) == msg
 
     assert info["coro"] is not None
@@ -85,7 +85,7 @@ def test_can_replace_the_loop(fut: asyncio.Future[None]) -> None:
 
     info["original"] = get_event_loop()
 
-    with OverrideLoop() as custom_loop:
+    with Loop() as custom_loop:
         futs["b"] = get_event_loop().create_future()
 
         info["before_run"] = get_event_loop()
@@ -115,7 +115,7 @@ def test_sets_None(self) -> None:
 
         original = get_event_loop()
 
-        with OverrideLoop(new_loop=False) as custom_loop:
+        with Loop(new_loop=False) as custom_loop:
             msg = "There is no current event loop in thread.*"
             try:
                 assert get_event_loop() is None
@@ -133,7 +133,7 @@ async def blah() -> None:
                 custom_loop.run_until_complete(coro)
                 assert False, "should have risen an error"
             except Exception as error:
-                msg = "OverrideLoop is not managing your overridden loop, use run_until_complete on that loop instead"
+                msg = "This alt_pytest_asyncio.Loop is not managing your overridden loop, use run_until_complete on that loop instead"
                 assert str(error) == msg
 
         assert get_event_loop() is original
@@ -179,13 +179,13 @@ async def outside1() -> None:
     assert info1 == [1, 2]
 
     # The way python asyncio works
-    # Means that by defining this outside our OverrideLoop
+    # Means that by defining this outside our Loop
     # The weakref held against it in the _asyncgens set on the loop
     # Will remain so that our shutdown_asyncgens function may work
     ag = my_generator(info2)
 
-    with OverrideLoop(new_loop=True) as custom_loop:
-        assert custom_loop.loop is not None
+    with Loop(new_loop=True) as custom_loop:
+        assert custom_loop.controlled_loop is not None
 
         assert info2 == []
         assert info3 == []
@@ -201,7 +201,7 @@ async def doit() -> None:
             assert info3 == [1]
 
         custom_loop.run_until_complete(doit())
-        assert list(custom_loop.loop._asyncgens) == [ag]  # type: ignore[attr-defined]
+        assert list(custom_loop.controlled_loop._asyncgens) == [ag]  # type: ignore[attr-defined]
         assert info3 == [1]
         assert info2 == [1, 2]
 
@@ -224,13 +224,11 @@ async def outside2() -> None:
 
 class TestTestingAutoUse:
     @pytest.fixture(autouse=True)
-    def custom_loop(
-        self, loop_info: dict[str, asyncio.AbstractEventLoop]
-    ) -> Iterator[OverrideLoop]:
+    def custom_loop(self, loop_info: dict[str, asyncio.AbstractEventLoop]) -> Iterator[Loop]:
         assert not loop_info
         loop_info["original"] = get_event_loop()
 
-        with OverrideLoop() as custom_loop:
+        with Loop() as custom_loop:
             assert loop_info["original"] is not get_event_loop()
             yield custom_loop
 
@@ -267,7 +265,7 @@ def assertLoopInfo(
     @pytest.fixture(autouse=True)
     async def fix1(
         self,
-        custom_loop: OverrideLoop,
+        custom_loop: Loop,
         loop_info: dict[str, asyncio.AbstractEventLoop],
     ) -> None:
         loop_info["fix1"] = get_event_loop()
@@ -313,7 +311,7 @@ def test_works(
 
     def test_has_the_loop_on_custom_loop(
         self,
-        custom_loop: OverrideLoop,
+        custom_loop: Loop,
         fix1: None,
         fix2: None,
         fix3: None,
@@ -323,12 +321,12 @@ def test_has_the_loop_on_custom_loop(
     ) -> None:
         loop_info["test"] = get_event_loop()
         assert list(loop_info) == ["original", "fix1", "fix2", "fix3a", "fix4a", "fix5a", "test"]
-        assert custom_loop.loop is get_event_loop()
+        assert custom_loop.controlled_loop is get_event_loop()
         self.assertLoopInfo(loop_info, closed=False)
 
     def test_can_use_futures_from_fixtures(
         self,
-        custom_loop: OverrideLoop,
+        custom_loop: Loop,
         fix1: None,
         fix2: None,
         fix3: None,