pytest-asyncio has a closed event loop, but only when running all tests

I had to tweak @matthewpark319's answer a bit, but adding a session-scoped fixture in conftest.py worked.

import asyncio

import pytest


@pytest.fixture(scope="session")
def event_loop():
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
    yield loop
    loop.close()

If you're using pytest-asyncio, you may also need to edit your pyproject.toml and add:

[tool.pytest.ini_options]
asyncio_mode = "auto"

https://pypi.org/project/pytest-asyncio/

You can apparently override pytest-asyncio's version of the event loop fixture. They have it like this:

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

I have it like this:

@pytest.fixture
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    cleanly_shutdown(loop)

Or in other cases like this:

@pytest.fixture
def event_loop():
    yield asyncio.get_event_loop()

def pytest_sessionfinish(session, exitstatus):
    asyncio.get_event_loop().close()

These docs are very helpful: https://docs.pytest.org/en/latest/reference/reference.html?highlight=sessionfinish#pytest.hookspec.pytest_sessionfinish

I ran into a very similar, if not the same, problem. (Pytest failed with "loop closed" only if I ran all the tests). The solution for me was different, so I thought I'd post it.

At the top of my conftest.py file I had:

    @pytest.fixture(params=["asyncio"], scope="session")
    def anyio_backend(request):
        return request.param

    # AsyncClient comes from HTTPX and "app" is FastAPI
    @pytest.fixture(scope="session")
    async def client() -> AsyncIterator[AsyncClient]:
        async with AsyncClient(app=app, base_url="http://testserver") as client:
            yield client

If I ran tests in a single module at a time (or in VSCode's test plugin) everything went fine. But if I ran all my tests via pytest terminal command, everything would pass and then I would get... THIS:

__________________________________________________ ERROR at teardown of TestValidations.test_validate_url_with_malformed_or_altered_token[asyncio] ___________________________________________________

anyio_backend = 'asyncio', args = (), kwargs = {}, backend_name = 'asyncio', backend_options = {}, runner = <anyio._backends._asyncio.TestRunner object at 0x1049eaad0>

    def wrapper(*args, anyio_backend, **kwargs):  # type: ignore[no-untyped-def]
        backend_name, backend_options = extract_backend_and_options(anyio_backend)
        if has_backend_arg:
            kwargs["anyio_backend"] = anyio_backend
    
        with get_runner(backend_name, backend_options) as runner:
            if isasyncgenfunction(func):
>               yield from runner.run_asyncgen_fixture(func, kwargs)

../../Library/Caches/pypoetry/virtualenvs/ab-bJoznT_5-py3.11/lib/python3.11/site-packages/anyio/pytest_plugin.py:68: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../Library/Caches/pypoetry/virtualenvs/ab-bJoznT_5-py3.11/lib/python3.11/site-packages/anyio/_backends/_asyncio.py:2097: in run_asyncgen_fixture
    self._loop.run_until_complete(fixture_task)
../../.pyenv/versions/3.11.4/lib/python3.11/asyncio/base_events.py:628: in run_until_complete
    self._check_closed()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

    def _check_closed(self):
        if self._closed:
>           raise RuntimeError('Event loop is closed')
E           RuntimeError: Event loop is closed

../../.pyenv/versions/3.11.4/lib/python3.11/asyncio/base_events.py:519: RuntimeError

Ugh.

Solution

The problem was rooted in the session-scoped client. I changed it to "module" scope and all was fixed! No need to manually mess with event loops.

    @pytest.fixture(params=["asyncio"], scope="session")
    def anyio_backend(request):
        return request.param

    # AsyncClient comes from HTTPX and "app" is FastAPI
    @pytest.fixture(scope="module")
    async def client() -> AsyncIterator[AsyncClient]:
        async with AsyncClient(app=app, base_url="http://testserver") as client:
            yield client

I'm not sure if it's because I wasn't explicitly cleaning up after the yield or perhaps HTTPX was closing the loop before anyio. (I'd welcome some insight from anyone more experienced than I.)

For anyone encountering this issue in the future, I'd recommend considering fixture scopes - perhaps it will save you from needing to play with event loops.

To resolve the issue of running multiple @pytest.mark.asyncio tests raising RuntimeError('Event loop is closed'), you can run all tests within the same event loop by setting scope="session". This can be done by either setting scope="session" for each test individually (see answer),

@pytest.mark.asyncio(scope="session")
async def test_1():
    ...

@pytest.mark.asyncio(scope="session")
async def test_2():
    ...

or by marking all tests using a pytest_collection_modifyitems hook in your conftest.py file:

import pytest
from pytest_asyncio import is_async_test

def pytest_collection_modifyitems(items):
    pytest_asyncio_tests = (item for item in items if is_async_test(item))
    session_scope_marker = pytest.mark.asyncio(scope="session")
    for async_test in pytest_asyncio_tests:
        async_test.add_marker(session_scope_marker, append=False)

source: run_session_tests_in_same_loop

Adding my solution here, because it took me forever to figure out and none of the solutions worked (correctly so).

The problem for me was, I was using a test that depended on an async fixture and a regular fixture in the wrong order. The regular fixture was a starlette.testclient.TestClient with lifespan events: https://www.starlette.io/lifespan/#running-lifespan-in-tests

I assume this is due to the order in which fixtures are requested, and it may be reverse order during shutdown, but I couldn't find anything in the docs.

import pytest_asyncio
from starlette.testclient import TestClient

# a FastAPI application with an async shutdown handler, e.g.
# app = FastAPI(on_shutdown=[lambda _: await asyncio.sleep(1)])
from main import app

@pytest_asyncio.fixture
async def some_fixture():
    # creates the event loop implicitly
    ...

def client():
    with TestClient(app) as client:
        yield client

def test_something(client, some_fixture):
    # breaks, event loop is closed at client shutdown
    ...

def test_something_else(some_fixture, client):
    # works, event loop is closed after client shutdown
    ...

This can also happen if you are using a teardown fixture that does the cleanup before yield

E.g.

async def teardown():

  # cleanup stuff involving async
  ..

  yield

On the second test this will fail, since each test case gets a different async loop, and thus the teardown will try to access the loop of the previous test but that has already been destroyed.

The fix is to put the cleanup after yield

async def teardown():
  yield

  # cleanup stuff involving async
  ..

This ensures the cleanup happens in the same async loop

wanted to update the solution for latest versions (pytest~=8.2.0 httpx~=0.27.0, fastapi==0.95.0),

Note: I am not using unittest.Testcase for the class inheritance and keeping a fixture for the async cleanup and autouse=True for my usecase.

in conftest.py fixture we follow following code:

@pytest.fixture(scope="session")
def anyio_backend():
    return 'asyncio'


@pytest.fixture(scope="session")
async def client():
    async with AsyncClient(app=app, base_url="http://test") as client:
    yield client

moreover in your code you need to mark the async tests as

 @pytest.mark.anyio
    async def test_read_ping(self, client: AsyncClient):
    assert response.status_code == 200

feel free to modify and test for your specific test case.

After implementing two highest score solutions. I am still facing the problem.

I finally use asyncio to run the test, instead of using @pytest.mark.asyncio:

def test_query_database():
    async def query_database():
        ...

    loop = asyncio.get_event_loop()
    loop.run_until_complete(query_database())

For me, what worked was to use aiounittest and define my own get_event_loop as suggested there.