"RuntimeError: Event loop is closed" when using pytest-asyncio to test FastAPI routes

Question

I received the error

RuntimeError: Event loop is closed

each time I try to make more than one async call inside my test. I already tried to use all other suggestions from other Stack Overflow posts to rewrite the event_loop fixture but nothing works. I wonder what I'm missing?

Run test command:

python -m pytest tests/ --asyncio-mode=auto

requirements.txt

pytest==7.1.2
pytest-asyncio==0.18.3
pytest-html==3.1.1
pytest-metadata==2.0.1

test.py

async def test_user(test_client_fast_api):
    assert 200 == 200

    # works fine
    request_first = test_client_fast_api.post("/first_route")

    # recieve RuntimeError: Event loop is closed
    request_second = test_client_fast_api.post("/second_route")

conftest.py

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

Gino Mempin · Accepted Answer · 2023-04-05 08:02:46Z

15

Add a file conftest.py to the directory where the test script is placed.

And write the following code:

import pytest
from main import app
from httpx import AsyncClient

@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:
        print("Client is ready")
        yield client

And then use those fixtures in your own test code. For example, this is the real test code for my own project. You can change it to your own.

import pytest
from httpx import AsyncClient

@pytest.mark.anyio
async def test_run_not_exists_schedule(client: AsyncClient):
    response = await client.get("/schedule/list")
    assert response.status_code == 200
    schedules = response.json()["data"]["schedules"]
    schedules_exists = [i["id"] for i in schedules]
    not_exists_id = max(schedules_exists) + 1
    request_body = {"id": not_exists_id}
    response = await client.put("/schedule/run_cycle", data=request_body)
    assert response.status_code != 200  

@pytest.mark.anyio
async def test_run_adfasdfw(client: AsyncClient):
    response = await client.get("/schedule/list")
    assert response.status_code == 200
    schedules = response.json()["data"]["schedules"]
    schedules_exists = [i["id"] for i in schedules]
    not_exists_id = max(schedules_exists) + 1
    request_body = {"id": not_exists_id}
    response = await client.put("/schedule/run_cycle", data=request_body)
    assert response.status_code != 200

Finally, run in the project's terminal

python -m pytest

If all goes well, it should be OK.

This may involve libraries that need to be installed.

pytest
httpx

edited Apr 5, 2023 at 8:02

Gino Mempin

30.5k31 gold badges125 silver badges174 bronze badges

answered Sep 22, 2022 at 8:58

Bai Jinge

3012 silver badges8 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

Gino Mempin Over a year ago

In case this answer isn't enough, try including this other answer: stackoverflow.com/a/73019163/2745495. The combination of both answers worked for me.

Anton Makarov Over a year ago

Tried your solution, but got Redirect (307) instead of 201 in testing fastapi-users route "/auth/register/"

Gino Mempin · Accepted Answer · 2023-04-05 13:37:06Z

5

This is the event loop fixture and TestClient pattern that worked for me:

from asyncio import get_event_loop
from unittest import TestCase

from async_asgi_testclient import TestClient

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

@pytest.mark.asyncio
    async def test_example_test_case(self):
        async with TestClient(app) as async_client:
            response = await async_client.get(
                "/api/v1/example",
                query_string={"example": "param"},
            )
        assert response.status_code == HTTP_200_OK

Refer to the relevant GitHub issue: https://github.com/tiangolo/fastapi/issues/2006#issuecomment-689611040

Please note - I could NOT figure our how to use Class based tests. Neither unittest.TestCase or asynctest.case.TestCase would work for me. pytest-asyncio docs (here) state that:

Test classes subclassing the standard unittest library are not supported, users are recommended to use unittest.IsolatedAsyncioTestCase or an async framework such as asynctest.

edited Apr 5, 2023 at 13:37

Gino Mempin

30.5k31 gold badges125 silver badges174 bronze badges

answered Oct 6, 2022 at 6:11

wjkw1

1741 silver badge7 bronze badges

1 Comment

Anton Makarov Over a year ago

Tried this one, but testing fastapi-users POST-routes "/auth/jwt/login/" and "/auth/register/" I get "405 Method not allowed" errors. While docs clearly show, these routes receive POST-requests. Confused.

John Snow · Accepted Answer · 2024-01-12 21:36:25Z

2

As of the latest pytest-asyncio==0.23.2 the issue can be fixed with

@pytest.mark.asyncio(scope="session")

Note that following wjkw1's answer, one will see the deprecation notice

  Replacing the event_loop fixture with a custom implementation is deprecated
  and will lead to errors in the future.
  If you want to request an asyncio event loop with a scope other than function
  scope, use the "scope" argument to the asyncio mark when marking the tests.
  If you want to return different types of event loops, use the event_loop_policy
  fixture.

answered Jan 12, 2024 at 21:36

John Snow

2,0482 gold badges30 silver badges50 bronze badges

4 Comments

Christof Kälin Over a year ago

True, but I still need my own event_loop fixture having pytest-asyncio==0.23.7. I tried any combination of the solutions suggested here and in other threads: scope="session" in the test annotations break every test and just sorting the fixtures (async-sync) did not change a thing (why should it?)

Pedro Werneck Over a year ago

For anyone still having this problem, if your application is using singletons for things like DB connection make sure you reset them between tests. I just spent a few hours debugging this and found out all tests after the first were trying to reuse a singleton DB client that was attached to the event loop of the first test, which had already been closed.

GoldenJoe Over a year ago

@PedroWerneck Just had the same problem myself. The 'scope' argument does fix it, but the root is the singleton DB client (in my case Firestore). Thanks for sharing.

Dave Ankin Jan 4 at 21:31

should be loop_scope="session" now

Roy Hyunjin Han · Accepted Answer · 2024-11-22 21:10:39Z

Set the default anyio backend
Use @pytest.fixture(scope='session') for your fixtures
Use @pytest.mark.anyio for your tests.

import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient

from your_application.applications import main

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

@pytest.fixture(scope='session')
async def client(app):
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url=base_uri,
    ) as client:
        yield client

@pytest.fixture(scope='session')
async def app():
    async with LifespanManager(main) as manager:
        yield manager.app

@pytest.mark.anyio
async def test_see_root_html(client):
    uri = main.url_path_for('see_root_html')
    response = await client.get(uri)
    assert response.status_code == 200

Reference: https://stackoverflow.com/a/72996947/192092

Collectives™ on Stack Overflow

"RuntimeError: Event loop is closed" when using pytest-asyncio to test FastAPI routes

4 Answers 4

2 Comments

1 Comment

4 Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

4 Answers 4

2 Comments

1 Comment

4 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related