Incompatible with anyio due to event loop keep-alive for session fixtures

[gaborbernat]

anyio creates an event loop during the fixture, and that loop will be reused/kept alive until the fixture is cleaned up (see https://anyio.readthedocs.io/en/stable/testing.html#technical-details). This means that if you have a session scooped fixture (which is quite common, as you might have a DB setup fixture):

@pytest.fixture(scope="session", name="db")
async def _setup_db() -> None:
    if await database_exists(ENGINE.url):  # pragma: no branch
        await drop_database(ENGINE.url)  # pragma: no cover
    await setup_db()

Then this fixture will keep alive the event loop until the end of the session (such fixture requires all your tests using the DB to reuse the same event loop).

pytest-alembic, however, assumes that when its test is running, there is no other event loop running, as it is calling

pytest-alembic/src/pytest_alembic/executor.py

Line 182 in 5ff6ca0

return asyncio.run(run(self.connection))

which is only allowed to be called when there is no other live event loop.

Instead, pytest-alembic IMHO should check if there is a live event loop and reuse it if so. A working workaround for now would be to create a new thread (that by default has no running loop on it):

    with ThreadPoolExecutor() as thread_pool:
        thread_pool.submit(upgrade, alembic_runner).result()

[DanCardin]

I almost wonder if the thread pool option would be preferable to trying to attach to the running loop. dealing with pytest-asyncio has already been a nightmare, and i definitely feel like I've tried to make it work more integratedly before, and it ran into many (potentially pytest-asyncio specific) issues.

[gaborbernat]

I almost wonder if the thread pool option would be preferable to trying to attach to the running loop. dealing with pytest-asyncio has already been a nightmare, and i definitely feel like I've tried to make it work more integratedly before, and it ran into many (potentially pytest-asyncio specific) issues.

yeah, that could be a good solution

[DanielHabenicht]

@gaborbernat Do you know where to (re)place code if you want to run pytest --test-alembic?

 with ThreadPoolExecutor() as thread_pool:
        thread_pool.submit(upgrade, alembic_runner).result()

[gaborbernat]

I never use that flag, so can't tell.

[DanCardin]

I am struggling to come up with a setup that works enough to produce the failing behavior. Ideally one of you could make a branch that copies or modifies, say, examples/test_async_sqlalchemy (pytest tests/test_runner.py::test_async_sqlalchemy) to the point where it fails in the right way? Or else a much more comprehensive set of code blocks here that enumerate a minimal conftest.py/fixtures setup and a minimal env.py.

With my naive attempt at a new test for asyncio (after uninstalling pytest-asyncio, because it appears to not play wel when they're installed together), it either fully succeeds or fails inside of env.py setup, not the actual tests.

I'll have to look into what precisely anyio is doing, but it seems like the test definitions need to be async functions in order for anyio to do anything with them; and to me that means the default --test-alembic is never going to work

which, i assume, means

async def test_upgrade(alembic):
    tests.test_upgrade(alembic)

only way that will work. I suppose perhaps a --async-test-alembic might be the solution there.