How to use the same event loop for fixtures and tests?

[rlubke]

Python 3.11.2
pytest-asyncio 0.24 (I've tried rolling back to 0.21, but no luck)

I'm writing integration tests for a library we publish that uses asyncio.

We have async fixtures that establish a Session (which is really a gRPC connection) to a backend, obtain a structure to dispatch various operations. This structure is then passed to the each of the async test functions.

If I run tests individually, everything is fine. However, if I run all of the tests in a particular test file, the first test passes and all subsequent tests fail in a fashion similar to:

ERROR                                  [  7%]2024-09-30 12:22:03,443 - coherence - INFO - Session [f3922653-17ff-4f2a-aae8-5a1a4e203350] connected to [localhost:1408].
### DEBUG USING V1
2024-09-30 12:22:03,445 - asyncio - ERROR - Exception in callback PollerCompletionQueue._handle_events(<_UnixSelecto...e debug=False>)()
handle: <Handle PollerCompletionQueue._handle_events(<_UnixSelecto...e debug=False>)()>
Traceback (most recent call last):
  File "/Users/rlubke/.pyenv/versions/3.11.3/lib/python3.11/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pyx.pxi", line 170, in grpc._cython.cygrpc.PollerCompletionQueue._handle_events
  File "/Users/rlubke/.pyenv/versions/3.11.3/lib/python3.11/asyncio/base_events.py", line 806, in call_soon_threadsafe
    self._check_closed()
  File "/Users/rlubke/.pyenv/versions/3.11.3/lib/python3.11/asyncio/base_events.py", line 519, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
2024-09-30 12:22:03,446 - asyncio - ERROR - Exception in callback PollerCompletionQueue._handle_events(<_UnixSelecto...e debug=False>)()
handle: <Handle PollerCompletionQueue._handle_events(<_UnixSelecto...e debug=False>)()>
Traceback (most recent call last):
  File "/Users/rlubke/.pyenv/versions/3.11.3/lib/python3.11/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "src/python/grpcio/grpc/_cython/_cygrpc/aio/completion_queue.pyx.pxi", line 170, in grpc._cython.cygrpc.PollerCompletionQueue._handle_events
  File "/Users/rlubke/.pyenv/versions/3.11.3/lib/python3.11/asyncio/base_events.py", line 806, in call_soon_threadsafe
    self._check_closed()
  File "/Users/rlubke/.pyenv/versions/3.11.3/lib/python3.11/asyncio/base_events.py", line 519, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

test setup failed
@pytest_asyncio.fixture
    async def setup_and_teardown() -> AsyncGenerator[NamedCache[Any, Any], None]:
        session: Session = await tests.get_session()
    
>       cache: NamedCache[Any, Any] = await session.get_cache("test")

test_client.py:47: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../src/coherence/client.py:114: in inner_async
    return await func(self, *args, **kwargs)
../src/coherence/client.py:1744: in get_cache
    await c.ensure_cache()
../src/coherence/client.py:95: in inner_async
    return await func(self, *args, **kwargs)
../src/coherence/client.py:930: in ensure_cache
    await dispatcher.dispatch(self._stream_handler)
../src/coherence/util.py:232: in dispatch
    await stream_handler.send_proxy_request(self._request)
../src/coherence/client.py:2296: in send_proxy_request
    await self._stream.write(proxy_request)
../../../../Library/Caches/pypoetry/virtualenvs/coherence-NZNGCFtv-py3.11/lib/python3.11/site-packages/grpc/aio/_call.py:470: in write
    await self._write(request)
../../../../Library/Caches/pypoetry/virtualenvs/coherence-NZNGCFtv-py3.11/lib/python3.11/site-packages/grpc/aio/_call.py:446: in _write
    await self._cython_call.send_serialized_message(serialized_request)
src/python/grpcio/grpc/_cython/_cygrpc/aio/call.pyx.pxi:371: in send_serialized_message
    ???
src/python/grpcio/grpc/_cython/_cygrpc/aio/callback_common.pyx.pxi:148: in _send_message
    ???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

>   ???
E   RuntimeError: Task <Task pending name='Task-16' coro=<_wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.setup() running at /Users/rlubke/Library/Caches/pypoetry/virtualenvs/coherence-NZNGCFtv-py3.11/lib/python3.11/site-packages/pytest_asyncio/plugin.py:280> cb=[_run_until_complete_cb() at /Users/rlubke/.pyenv/versions/3.11.3/lib/python3.11/asyncio/base_events.py:180]> got Future <Future pending> attached to a different loop

src/python/grpcio/grpc/_cython/_cygrpc/aio/callback_common.pyx.pxi:99: RuntimeError

I've tried setting the loop_scope to session for each test without any luck.
I've tried using conftest.py approach as outlined here but this approach fails with:

ImportError: cannot import name 'is_async_test' from 'pytest_asyncio' (/Users/rlubke/Library/Caches/pypoetry/virtualenvs/coherence-NZNGCFtv-py3.11/lib/python3.11/site-packages/pytest_asyncio/__init__.py)

I'm running out of ideas on how to approach/further diagnose this (I'm primarily a Java developer working on Python stuff on the side, so bear with me). Any pointers on how I can tackle getting to the bottom of this?

Some sample code from our tests:

@pytest_asyncio.fixture
async def setup_and_teardown() -> AsyncGenerator[NamedCache[Any, Any], None]:
    session: Session = await tests.get_session()

    cache: NamedCache[Any, Any] = await session.get_cache("test")

    yield cache  # this is what is returned to the test functions

    await cache.truncate()
    await session.close()

@pytest.mark.asyncio(loop_scope="session")
async def test_put_with_ttl(setup_and_teardown: NamedCache[str, Union[str, int]]) -> None:
    cache: NamedCache[str, Union[str, int, Person]] = setup_and_teardown

    k: str = "one"
    v: str = "only-one"
    await cache.put(k, v, 5000)  # TTL of 5 seconds
    r = await cache.get(k)
    assert r == v

    sleep(5)  # sleep for 5 seconds
    r = await cache.get(k)
    assert r is None

[WisdomPill]

any update on how to solve this? as of now the only solution I've found is to override event_loop fixture which is discouraged.

in conftest.py

@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

[seifertm]

@WisdomPill To run all fixtures in the same event loop, you can set asyncio_default_fixture_loop_scope. This how-to guide describes a way to run all tests in the same loop.

[domartynov]

@WisdomPill To run all fixtures in the same event loop, you can set asyncio_default_fixture_loop_scope. This how-to guide describes a way to run all tests in the same loop.

@seifertm the suggested solution only makes all fixtures to share the same event loop, tests get separate instances (or an instance if the session test scope is used). What was the reason to deprecate the event_loop fixture?

[seifertm]

@domartynov The next release will also add a similar option for tests (#793)
Let me know if #706 (comment) explains the reason for the deprecation well enough.