Question: how to set fixture and async function same loop?

[Provinm]

here is my code

test_something.py

@pytest_asyncio.fixture(scope="session")
async def setup_env():
    import asyncio

    loop1 = asyncio.get_running_loop()
    print(f"setup_env {loop1} - {id(loop1)}")
    
    yield 




async def test_endpoints(setup_env: t.Generator) -> None:
    import asyncio

    loop2 = asyncio.get_running_loop()
    print(f"test_endpoints {loop2} - {id(loop2)}")

and conftest.py

def pytest_collection_modifyitems(items: t.List[Item]) -> None:
    pytest_asyncio_tests = (item for item in items if is_async_test(item))
    session_scope_marker = pytest.mark.asyncio(loop_scope="session")
    for async_test in pytest_asyncio_tests:
        async_test.add_marker(session_scope_marker, append=False)

my question is how to make loop1 == loop2

[Provinm]

solved, thx!

[scastlara]

How did you solve it?

EDIT: For me it was adding this to pyproject.toml

[tool.pytest.ini_options]
asyncio_default_fixture_loop_scope = "session"

And following the docs and adding this to conftest.py: https://pytest-asyncio.readthedocs.io/en/latest/how-to-guides/run_session_tests_in_same_loop.html

That way all the tests run on the same loop as the fixtures (I have session-scoped db fixture that runs migrations, and a function scoped db fixture that wraps everything in a transaction; so I needed everything to run on the same loop).

[Provinm]

@scastlara

change

@pytest_asyncio.fixture(scope="session")
async def setup_env():
    import asyncio

    loop1 = asyncio.get_running_loop()
    print(f"setup_env {loop1} - {id(loop1)}")
    
    yield 

to

@pytest_asyncio.fixture(loop_scope="session")
async def setup_env():
    import asyncio

    loop1 = asyncio.get_running_loop()
    print(f"setup_env {loop1} - {id(loop1)}")
    
    yield 

this makes loop1 == lop2

[nekeal]

@scastlara, have you encountered any issues with this approach? I have a similar case:

• A session-scoped fixture for AsyncEngine from SQLAlchemy
• A function-scoped fixture for AsyncSession
• Setting asyncio_default_fixture_loop_scope = "session"

I'm curious if the approach you mentioned is the best way to handle this, or if there might be a better alternative.

[scastlara]

@scastlara, have you encountered any issues with this approach? I have a similar case:

• A session-scoped fixture for AsyncEngine from SQLAlchemy
• A function-scoped fixture for AsyncSession
• Setting asyncio_default_fixture_loop_scope = "session"

I'm curious if the approach you mentioned is the best way to handle this, or if there might be a better alternative.

Hey @nekeal ! No problems at all.

• In pyproject.toml

asyncio_default_fixture_loop_scope = "session"

• In conftest.py

def pytest_collection_modifyitems(items: Any) -> None:
    """
    Make all tests run on the same event loop.

    This is necessary because our db session is created on the event loop,
    and we need to ensure all tests run on the same event loop as the db!
    See: https://pytest-asyncio.readthedocs.io/en/latest/how-to-guides/run_session_tests_in_same_loop.html
    """
    pytest_asyncio_tests = (item for item in items if is_async_test(item))
    session_scope_marker = pytest.mark.asyncio(loop_scope="session")
    for async_test in pytest_asyncio_tests:
        async_test.add_marker(session_scope_marker, append=False)



@pytest.fixture(scope="session", autouse=True)
async def prepare_testsuite_db() -> AsyncIterator[None]:
    """Set up test database and engine."""
    engine = create_async_engine(get_database_url())

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
        await conn.commit()

        yield

# Then the func scope db fixtures providing a transaction for each test

The testsuite is just as fast as before, no errors at all. Only thing lacking is being able to configure the loop scope of tests in pyproject.toml instead of having to do tricks with pytest hooks. But I think there was an open issue about this and the lib maintainer is fully aware. For now this is sufficient for me.

[nekeal]

Thank you, @scastlara! I agree that the hook is not an ideal solution, but just to be thorough, I ran a small benchmark. For a test suite consisting of 1,100 synchronous and 820 asynchronous tests, it takes less than a millisecond. So that's a good enough solution 🎉