agronholm / anyio

High level asynchronous concurrency and networking framework that works on top of either trio or asyncio
MIT License
1.78k stars 135 forks source link

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

Open gaborbernat opened 2 days ago

gaborbernat commented 2 days ago

Things to check first

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 https://github.com/schireson/pytest-alembic/issues/119 when debugging this)

agronholm commented 23 hours ago

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 commented 17 hours ago

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.