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

Answer 1

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

Answer 2

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.