Sync methods do not trigger async fixtures even if pytest.mark.anyio annotated

[gaborbernat]

Things to check first

• I have searched the existing issues and didn't find my bug already reported there

• I have checked that my bug is still present in the latest release

AnyIO version

4.6.0

Python version

3.12

What happened?

When trying to use this library with https://github.com/schireson/pytest-alembic (that has sync tests calling the async code https://pytest-alembic.readthedocs.io/en/latest/asyncio.html) I noticed that if you have an async fixture for a sync test, the fixture is not evaluated and instead the coroutine is inserted as value for the fixture.

How can we reproduce the bug?

from pytest_alembic import Config, MigrationContext
from pytest_alembic.tests import test_upgrade as upgrade

@pytest.fixture(scope="session")  # must be session scoped to support sessions scooped DB setup
def anyio_backend() -> tuple[str, dict[str, bool]]:
    return "asyncio", {"use_uvloop": True}


@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()


@pytest.mark.usefixtures("db")
async def test_upgrade(alembic_runner: MigrationContext) -> None:  # noqa: RUF029 # fixture is async
   upgrade(alembic_runner)

(Note I also ran into schireson/pytest-alembic#119 when debugging this)

[agronholm]

There is nothing in that test that would cause it to be picked up by the AnyIO pytest plugin. You have an anyio_backend fixture there but it's not used by either the fixture or the test. Try adding pytestmark = pytest.mark.anyio.

Furthermore, the description speaks of sync tests, but in the snippet you have an async test. Is this just a copy/paste error?

Lastly, the pytest plugin is intended for running async test functions, and won't activate for sync test functions. I've been thinking of extending the functionality to run async fixtures on sync functions, but I didn't have a plausible use case for it. Additionally, given that the event loop would be suspended during a sync test function, async fixtures could only be used to generate test data, and not provide background services (which is what I usually use them for). It could be a footgun causing people to complain that their tasks hang while the test is running.

[gaborbernat]

My example is flawed, but yes, I have the marker and yes, the async for test_upgrade is what I need to do now to make it working. Here is the fixed repro:

from pytest_alembic import Config, MigrationContext
from pytest_alembic.tests import test_upgrade as upgrade

pytestmark = pytest.mark.anyio

@pytest.fixture(scope="session")  # must be session scoped to support sessions scooped DB setup
def anyio_backend() -> tuple[str, dict[str, bool]]:
    return "asyncio", {"use_uvloop": True}


@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()


@pytest.mark.usefixtures("db")
def test_upgrade(alembic_runner: MigrationContext) -> None: 
   upgrade(alembic_runner)

Lastly, the pytest plugin is intended for running async test functions, and won't activate for sync test functions.

Yes, this is my problem at this time.

I've been thinking of extending the functionality to run async fixtures on sync functions, but I didn't have a plausible use case for it.

Providing the use case right in this issue 😊

Additionally, given that the event loop would be suspended during a sync test function, async fixtures could only be used to generate test data, and not provide background services (which is what I usually use them for). It could be a footgun causing people to complain that their tasks hang while the test is running.

I could live with a world where this would raise an error out of the box (rather than silently pass through the coroutine), and require a pytest.mark.anyio_async_fixutre or something so that the user explicitly opts in, or make it a plugin config.

[agronholm]

I could live with a world where this would raise an error out of the box (rather than silently pass through the coroutine), and require a pytest.mark.anyio_async_fixutre or something so that the user explicitly opts in, or make it a plugin config.

Raise an error, under what conditions? If the anyio marker is applied to a sync test function? That would break just about every single test suite I have right now since I use pytestmark = pytest.mark.anyio.

[jakkdl]

I could live with a world where this would raise an error out of the box (rather than silently pass through the coroutine), and require a pytest.mark.anyio_async_fixutre or something so that the user explicitly opts in, or make it a plugin config.

Raise an error, under what conditions? If the anyio marker is applied to a sync test function? That would break just about every single test suite I have right now since I use pytestmark = pytest.mark.anyio.

This is #789 / pytest-dev/pytest#10839, no?

[agronholm]

Are you suggesting that the anyio pytest plugin should be triggered for sync functions, but then instead of running the function in question, it should check if it depends on async fixtures and raise an error then?

[jakkdl]

Are you suggesting that the anyio pytest plugin should be triggered for sync functions, but then instead of running the function in question, it should check if it depends on async fixtures and raise an error then?

#789 suggests directly erroring if the anyio marker is applied to a sync test, this is the current functionality in pytest-trio. But only erroring if it depends on async fixtures (i.e. your interpretation of my suggestion) would avoid the issue of breaking test suites that use pytestmark = pytest.mark.anyio in files with sync tests.

For @gaborbernat I think you should simply make the test async, even if it only contains sync code in the test body, since it's pretty much equivalent to

async def test_upgrade():
  async with CreateAlembicRunner() as alembic_runner:
    upgrade(alembic_runner)

[jakkdl]

Okay scrap my previous suggestions, pytest-dev/pytest#12930 will make unhandled async fixtures warn/error, fixing #789, so the only remaining question is if we want this to work by changing

anyio/src/anyio/pytest_plugin.py

Lines 126 to 133 in bdf09a6

	def pytest_pycollect_makeitem(collector: Any, name: Any, obj: Any) -> None:
	if collector.istestfunction(obj, name):
	inner_func = obj.hypothesis.inner_test if hasattr(obj, "hypothesis") else obj
	if iscoroutinefunction(inner_func):
	marker = collector.get_closest_marker("anyio")
	own_markers = getattr(obj, "pytestmark", ())
	if marker or any(marker.name == "anyio" for marker in own_markers):
	pytest.mark.usefixtures("anyio_backend")(obj)

to not just trigger if iscoroutinefunction(...) but always adding anyio_backend to the sync test.

And @gaborbernat you can keep your test function sync if you add anyio_backend as a fixture to the test function or the fixture.

[EmmanuelNiyonshuti]

Now that pytest-dev/pytest#12930 landed a good a while back and #789 has been fixed.
on using async fixtures in sync test, If I am interpreting the suggestion correctly, it's proposing to insert anyio_backend into every test regardless of whether it's sync or async. Isn't that quite implicit and intrusive for tests that have nothing to do with async code?

If you want to depend on an async fixture from a sync test, you can already explicitly add the anyio_backend fixture to your async fixture. Perhaps we could just document something like that. And for anyone coming across this issue: depending on async fixtures from a sync test would be useful if the fixture is generating your data (fetched data or something similar). I put together this repro to show why it's not something you'd want for background-task fixtures:

import pytest


progress: list[float] = []

@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"

@pytest.fixture
async def background_ticker(anyio_backend: str) -> None:
    async def tick() -> None:
        while True:
            progress.append(time.monotonic())
            await anyio.sleep(0.02)
            print("Hello")

    async with anyio.create_task_group() as tg:
        tg.start_soon(tick)
        yield
        tg.cancel_scope.cancel()

def test_sync_body_gets_no_ticks(background_ticker) -> None:
    progress.clear()

    start = time.monotonic()
    time.sleep(0.5)
    elapsed = time.monotonic() - start

    ticks_during_sync_body = len(progress)
    print(
        f"\nelapsed sync sleep: {elapsed:.3f}s | "
        f"ticks recorded during that window: {ticks_during_sync_body}"
    )

    assert ticks_during_sync_body == 0, (
        f"expected 0 ticks during the sync body, "
        f"got {ticks_during_sync_body}"
    )

@pytest.mark.anyio
async def test_async_body_gets_ticks(background_ticker) -> None:
    progress.clear()
    await anyio.sleep(0.5)
    ticks_during_async_body = len(progress)
    print(f"\nticks recorded during async body: {ticks_during_async_body}")
    assert ticks_during_async_body > 0