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

Question

I have a test to verify an exception is thrown from an async response, which I'm using pytest-asyncio version 0.10.0 to run.

Code is basically:

class TestThis:
    @pytest.mark.asyncio
    def test_the_thing(self):
       arg1 = "cmd"
       arg2 = "second command"
       with pytest.raises(CustomException):
           await do_thing(arg1, arg2)

Now the really weird thing is this test works fine if I run it alone, or if I run the class alone. However when I run all tests (pytest at the project root), it fails every time with a runtime error, saying the loop is closed.

Certes

Do you close the loop in another test? – Rarefied

Please provide a minimal reproducible example. – Halfcock

@Rarefied - no, this is the only async test in the suite – Certes

So, not necessarily an answer, but the test was living in a subfolder of the tests folder Moving the test from services to the root tests, it now works? – Certes

This might help you: github.com/pytest-dev/pytest-asyncio/issues/30 – Salot

Answer 1

There is an update available after the pytest-asyncio>=0.23.0 release.

After that we can have multiple asyncio tests.

import pytest

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

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

Magdeburg

by default, scope is set to function. setting scope to session may result in unintended side effects and flaky tests, as these tests are no longer isolated – Alexine

Answer 2

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"

Incrust

loop = asyncio.get_event_loop() yields a DeprecationWarning: There is no current event loop on Python 3.10. docs.python.org/3.10/library/… – Woe

I do not know why, but this solution works only not for all my projects. – Eliathan

I logged just to upvote this one :) – Bazil

Why isn't this a bug in pytest-asyncio? – Catkin

Answer 3

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

Havoc

Answer 4

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.

Praxiteles

Answer 5

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

Cioban

Answer 6

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

Amaze

Answer 7

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

Vittle

Answer 8

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.

Thais

Answer 9

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
    ...

Hilmahilt

Answer 10

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

Alexine