Design discussion: support higher-scoped (class/module/session level) fixtures?

[oremanj]

I'm pretty sure it's possible if we're willing to use greenlet.

Here's a sketch that's missing important details like exception propagation.

Hook pytest_runtest_loop to keep a Trio event loop running in another greenlet for the whole pytest invocation:

async def _manage_fixture(teardown_evt, done_evt, coroutine_or_agen, *, task_status):
    teardown = False
    if not hasattr(coroutine_or_agen, "asend"):
        # fixture with a return statement
        result = await coroutine_or_agen
    else:
        # fixture with a yield statement
        result = await coroutine_or_agen.asend(None)
        teardown = True
    task_status.started(result)
    if teardown:
        try:
            await coroutine_or_agen.asend(None)
        except StopAsyncIteration:
            pass
        else:
            raise RuntimeError("too many yields in fixture")
    done_evt.set()

async def _pytest_trio_main(pytest_greenlet):
    async with trio.open_nursery() as nursery:
        fixtures = {}
        next_msg = "ready"
        while True:
            request = pytest_greenlet.switch(next_msg)
            if request[0] == "setup":
                fixtures[request[1]] = (teardown_evt, done_evt) = (trio.Event(), trio.Event())
                next_switch = await nursery.start(_manage_fixture, teardown_evt, done_evt, request[1])
            elif request[0] == "teardown":
                teardown_evt, done_evt = fixtures.pop(request[1])
                teardown_evt.set()
                await done_evt.wait()
                next_switch = None
            elif request[0] == "run":
                next_switch = await request[1]
            else:
                assert request[0] == "exit"
                return "done"

@pytest.hookimpl(hookwrapper=True)
def pytest_runtestloop(session):
    pytest_greenlet = greenlet.getcurrent()
    session._pytest_trio_trio_greenlet = pytest_greenlet  # or somewhere better
    trio_greenlet = greenlet.greenlet(trio.run)
    first_send = trio_greenlet.switch(_pytest_trio_main, pytest_greenlet)
    assert first_send == "ready"
    yield
    del session._pytest_trio_trio_greenlet
    assert trio_greenlet.switch("exit") == "done"

Hook fixture setup to run async fixtures in the Trio greenlet:

@pytest.hookimpl(hookwrapper=True)
def pytest_fixture_setup(fixturedef, request):
    wrapped_result = yield
    result = wrapped_result.get_result()
    if not hasattr(result, "__await__") and not hasattr(result, "__aiter__"):
        return
    trio_greenlet = request.session._pytest_trio_trio_greenlet
    true_result = trio_greenlet.switch("setup", result)
    if hasattr(result, "__aiter__"):
        request.addfinalizer(partial(trio_greenlet.switch, "teardown", result))
    wrapped_result.force_result(true_result)
    my_cache_key = fixturedef.cache_key(request)
    fixturedef.cached_result[my_cache_key] = (true_result, my_cache_key, None)

Hook running-the-test to run async tests in the Trio greenlet (not shown).

The details are decidedly nontrivial, but I don't see any showstoppers... thoughts? Maybe this could totally replace pytest-trio's current strategy, but it's not a win across the board, so I'm imagining it more as an alternative mode.

Benefits:

• higher-scoped fixtures!
• can trivially take advantage of pytest's ordering of fixtures; no need for most of the current tricky setup/teardown logic
• could imagine running multiple tests in parallel, since you have the shared event loop all set up -- this would involve setting up multiple greenlets to do pytest_runtest_protocol on different subsets of the session.items

Drawbacks:

• depends on greenlet (though PyPy does support it at least)
• if you don't propagate the exceptions totally right, can expect super confusing tracebacks
• can't use custom clocks/instruments per test
• running fixture setup in parallel is much harder

[touilleMan]

Using the same trio loop with different tests is a very slippy slope (typically with the current nursery fixture we can endup with a test spawning coroutines that will outlive it)

Adding greenlet to the mix is a whole new layer of complexity, pytest-trio is already far from trivial so this seems like another big red flag.

I think the best way to implement higher level fixture is to use them to start a separate trio loop within a thread.
The two isolated trio loops (one for the class/module/session fixture, one for the test) is a double edged sword: you have to do extra work if you want to synchronize the two loop, BUT it also mean you cannot break stuff due to subtle interactions between coroutines.
So I guess we shouldn't even add this feature to the pytest-trio library given the user has to understand the limitations of the class/module/session fixture. The best we could do would be to add a recipe in the documentation.

[tomprince]

I was investigating using pytest-trio to write some integration tests, where I have some expensive services to setup, so want to have session scoped fixtures for the services. I also want to collect logs from the services as they run, and sometimes restart those services from the tests. It is definitely less appealing if I need to manage my own loop for anything session scoped, or that interacts with code from the session scoped things.

In terms of implementation, it looks like how trio is structured to support guest mode could be used to support suspending and resuming the loop. With the current public API, you could have one long-running trio loop, which is what is exposed to users; then, when running a async test or fixture, iterate the host-loop calls (perhaps using a separate trio loop). This would require at least a little glue to capture the guest calls to run_sync_soon_threadsafe and pass them to the current host loop, or batch them until it starts. Or, with access to internals, or a new API, the loop could be suspended after each fixture or callback is completed.

It is definitely true that you can't use the current nursery fixture from non-function scoped fixtures; the same is true of pytest's temp_path fixture, which is why there is the temp_path_factory fixture for use from those.

[svermeulen]

I'm unclear on what the danger is in using the same trio loop for both session and module fixtures in addition to just function fixtures. I don't understand why we couldn't have a session-scope nursery, a module-scope nursery under that, then the function-level nursery under that, then automatically cancel the nurseries when the module is complete and at the end of the session, in the same way the function level nursery is already cancelled. Is it somehow possible for an async method to outlive the nursery it was spawned in?

[jmehnle]

I wanted to chime in here to emphasize that the lack of support for session/module/class-scoped async fixtures is a serious limitation for us. We're writing unit tests to exercise code accessing a PostgreSQL database using the asyncpg package, and we're trying to spin up a Docker container with PostgreSQL that the tests can run against. Obviously spinning up a Docker container and tearing it down isn't something you can afford to do for every single database test. In a previous life I used the synchronous psycopg package, and defining a session-scoped fixture to manage a database container worked beautifully, but now I've spent the last two days trying to find a way to work around this pytest-trio limitation, to no avail.

I know that pytest-trio advertises its ability to run async fixtures concurrently as a feature, but I would argue that if that feature had to be weighed against support for session-scoped fixtures, the benefit of session-scoped fixtures would almost certainly weigh a lot more heavily from any practical point of view given how much test time can be saved with setting up expensive resources only once vs. many times.

I might be willing to donate some effort to make session/module/class-scoped async fixtures in pytest-trio a reality, but despite digging rather deeply and reading the above discussion I still don't have a good understanding of what the fundamental blocker is (though I do agree that introducing a dependency on greenlet seems wrong). My vague notion is that pytest-trio effectively runs a fresh loop for every async test it handles, so there's no long-lived loop that session-scoped fixtures could setup under, be suspended (yielded) under, and ultimately tear down under. Can you help me understand this better?

[oremanj]

@jmehnle The basic difficulty is that all of pytest's internal test runner logic is synchronous code, so we can't easily open an event loop "around" the test runner and run tests inside of it, because synchronous functions can't do a blocking call into an enclosing async context. However, we now have two technologies that we didn't have when this issue was first opened, either of which might be able to bridge that gap:

• Trio guest mode allows one to start a Trio run "in the background" and then yield to run more synchronous code, which can start tasks in the background Trio run. You still need to implement a stub "host" event loop for Trio to be a guest of, but this can be a simple "execute callbacks until this event occurs" type of thing, and Trio's own tests of guest mode have some examples of how to do that. Under the guest mode approach, an async higher-scoped fixture's setup logic would do "start a task that runs the fixture, then pump callbacks until it reaches its yield" while teardown logic would be "set an event that causes the task to resume, then pump callbacks until it exits, then raise any exception that resulted". The plumbing will probably be annoying.
• My greenback package provides a more drop-in solution for the async-sync-async sandwich problem. This still incurs a dependency on greenlet, but you don't have to manage the greenlet(s) yourself.

Of these I think guest mode is probably a better fit for pytest, and it also avoids the greenlet dependency.

Higher-scoped fixtures require a single Trio run that wraps all tests in their scope. This is incompatible with current popular use of fixtures like autojump_clock, which require that each test have its own Trio run so some of them can use the autojump clock while others don't (the clock must be known when the run starts and can't be swapped out partway). So I think this might actually want to be a separate package from pytest-trio (maybe pytest-xtrio?). Or pytest-trio could support both approaches, but it would need to maintain almost completely independent code for each approach.

To solve your immediate problem, it is probably going to be much much easier to use a synchronous session-level fixture that starts a thread and runs the container from the thread.

[jmehnle]

Thanks for the great overview. I will ponder this in the coming weeks.

FWIW, and for anyone having a similar problem, I have worked around the limitation by using a synchronous session-scoped fixture that spins up the database container and talks to the database using a synchronous database driver to set up the DB, in deviation from the asynchronous driver we use in application code. This is pretty ugly, and I'd still very much love to get rid of this pile of dirt.

[bradleyharden]

Echoing @jmehnle's comment, we use pytest for hardware testing, and we use session-scoped fixtures for things like setting up and configuring the hardware. Our tests are very IO-bound and timeout-heavy, so I would really like to use trio instead of our current solution. But the scope issue is almost a non-starter.

That being said, I had an idea for an alternative architecture of pytest-trio that seems to solve the issue rather simply, yet I've seen no discussion of this architecture so far, including the original issue on the trio issue tracker.

Instead of having a per-test call to trio.run, why not run trio in a separate thread from pytest and run async functions on it using trio.lowlevel.from_thread.run. From pytest's perspective, everything is still synchronous, since it doesn't move on until the call to from_thread.run returns. That also moves all of the fixture management logic back into pytest, where it belongs.

This architecture seems to greatly simplify the plugin logic. So far, I can't think of any major downsides. I've provided a complete, working confest.py implementation of it below. Is there some reason this architecture was considered but rejected? Can @njsmith or @touilleMan comment?

from inspect import iscoroutinefunction, isasyncgenfunction
from math import inf
from queue import Queue
from threading import Thread

import pytest
from pytest import Session, StashKey, Item, FixtureDef, FixtureRequest
import trio
from trio import CancelScope
from trio.lowlevel import TrioToken


trio_token_key = StashKey[TrioToken]()
trio_thread_key = StashKey[Thread]()
trio_cancel_key = StashKey[CancelScope]()


type TrioQueue = Queue[tuple[TrioToken, CancelScope]]


async def trio_main(queue: TrioQueue) -> None:
    trio_token = trio.lowlevel.current_trio_token()
    with CancelScope() as cancel_scope:
        queue.put((trio_token, cancel_scope))
        await trio.sleep(inf)


@pytest.hookimpl(wrapper=True)
def pytest_sessionstart(session: Session):
    try:
        # If pytest.main was called with trio.lowlevel.to_thread.run_sync,
        # then a TrioToken should be available.
        trio_token = trio.lowlevel.current_trio_token()
    except RuntimeError:
        # Otherwise, start a dedicated trio thread
        queue: TrioQueue = Queue()
        trio_thread = Thread(target=trio.run, args=(trio_main, queue))
        trio_thread.start()
        trio_token, cancel_scope = queue.get()
        session.stash[trio_thread_key] = trio_thread
        session.stash[trio_cancel_key] = cancel_scope
    session.stash[trio_token_key] = trio_token
    yield


@pytest.hookimpl(wrapper=True)
def pytest_fixture_setup(fixturedef: FixtureDef, request: FixtureRequest):
    fn = fixturedef.func
    trio_token = request.session.stash[trio_token_key]
    if isasyncgenfunction(fn):
        def gen_adapter(*args, **kwargs):
            gen = fn(*args, **kwargs)
            yield trio.from_thread.run(anext, gen, trio_token=trio_token)
            try:
                trio.from_thread.run(anext, gen, trio_token=trio_token)
            except StopAsyncIteration:
                pass
        fixturedef.func = gen_adapter
    elif iscoroutinefunction(fn):
        def coro_adapter(*args, **kwargs):
            coro = fn(*args, **kwargs)
            async def inner():
                return await coro
            return trio.from_thread.run(inner, trio_token=trio_token)
        fixturedef.func = coro_adapter
    yield


@pytest.hookimpl(wrapper=True)
def pytest_runtest_call(item: Item):
    fn = item.obj
    if iscoroutinefunction(fn):
        trio_token = item.session.stash[trio_token_key]
        def adapter(*args, **kwargs):
            coro = fn(*args, **kwargs)
            async def inner():
                return await coro
            trio.from_thread.run(inner, trio_token=trio_token)
        item.obj = adapter
    yield


@pytest.hookimpl(wrapper=True)
def pytest_sessionfinish(session: Session, exitstatus):
    cancel_scope = session.stash.get(trio_cancel_key, None)
    if cancel_scope is not None:
        trio_token = session.stash[trio_token_key]
        trio_thread = session.stash[trio_thread_key]
        trio.from_thread.run_sync(cancel_scope.cancel, trio_token=trio_token)
        trio_thread.join()
    yield

[touilleMan]

Hi @bradleyharden ;-)

It's been a while since I've touched this pytest-trio so I won't be able to provide you with much information...

My guess regarding why pytest-trio never used a threading strategy is we never needed it: the first version of the plugin I wrote was simple enough, then we iterated (at lot) from there.

As you say, your approach seems appealing regarding its simplicity, however if I've learned something working with asynchronous codebase, it's things are always more complicated than first expected (and that's usually @njsmith's legendary wisdom that brings us the "more complicated part" 😄 )

btw have you had a look at #147 ? (given the issue is also about refactoring this plugin)

[bradleyharden]

@touilleMan, yes, I fully anticipated that it might be more complicated than I thought, hence my question of whether the architecture was considered and rejected for some reason.

In any event, I did see the other thread, but I didn't look into it closely. I think it's a better fit for this discussion.

[jakkdl]

Note that it's currently possible to get higher-scoped fixtures if you use the anyio pytest plugin, which you can configure to only run in trio mode. It doesn't have 100% feature parity with pytest-trio, but it's fairly close atm.