async test is RuntimeError: no running event loop in router with await

[wwstory]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from fastapi import FastAPI
import asyncio
import pytest
from httpx import AsyncClient


app = FastAPI()


async def query_db():
    await asyncio.sleep(0.1)    # add this, it's error
    return {'id': 1, 'data': 'hi'}


@app.get('/')
async def get_data():
    data = await query_db()
    return data


@pytest.mark.anyio
async def test_get_data():
    # async with AsyncClient(base_url='http://127.0.0.1:8000') as ac: # don't use `app=app`, need manual launch server, but it's ok.
    async with AsyncClient(app=app, base_url='http://127.0.0.1:8000') as ac:
        response = await ac.get('/')
    assert response.status_code == 200

Description

execute pytest in terminal
it is raise RuntimeError: no running event loop.

in fact, it is RuntimeError: Event loop is closed in my project.

Operating System

Linux

Operating System Details

ubuntu20.04

FastAPI Version

0.72.0

Python Version

3.8.10

Additional Context

No response

[blokje]

This is not a FastAPI issue but is related to AnyIO. More information about testing with AnyIO can be found at https://anyio.readthedocs.io/en/stable/testing.html.

In this case you are missing an instruction on which backend to use for testing. By adding the following snippet to your code (copied from link above) your test will run:

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

[wwstory]

@blokje Thank you for your reply. It does solve the above problem. Sorry, I'm not good at writing asynchronous code. But in fact, my code should be like this.

Example Code

from fastapi import FastAPI
import pytest
from httpx import AsyncClient
from motor.motor_asyncio import AsyncIOMotorClient


app = FastAPI()
client = AsyncIOMotorClient(f'mongodb://root:123456@127.0.0.1:27017/my_db')
db = client['my_db']


@app.get('/')
async def get_data():
    data = await db['users'].find().to_list(None)
    data = [{k:v for k,v in d.items() if k != '_id'} for d in data] # delete '_id' field
    return data

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

@pytest.mark.anyio
async def test_get_data():
    # async with AsyncClient(base_url='http://127.0.0.1:8000') as ac: # don't use `app=app`, need manual launch server, but it's ok.
    async with AsyncClient(app=app, base_url='http://127.0.0.1:8000') as ac:
        response = await ac.get('/')
    assert response.status_code == 200

Description

I get this error:
E RuntimeError: Task <Task pending name='Task-1' coro=<test_get_data() running at /root/workspace/demo/demo13/test_testing.py:26> cb=[_run_until_complete_cb() at /usr/lib/python3.8/asyncio/base_events.py:184]> got Future <Future pending> attached to a different loop

[CanD42]

there is an motor init error somewhere introduced with 0.69 and any.io, so this could be similar

you can try:
client.get_io_loop = asyncio.get_event_loop

I found this solution here: encode/starlette#1315

[Kludex]

from fastapi import FastAPI
import pytest
from httpx import AsyncClient
from motor.motor_asyncio import AsyncIOMotorClient


app = FastAPI()
client = AsyncIOMotorClient(f'mongodb://root:123456@127.0.0.1:27017/my_db')
db = client['my_db']


@app.get('/')
async def get_data():
    data = await db['users'].find().to_list(None)
    data = [{k:v for k,v in d.items() if k != '_id'} for d in data] # delete '_id' field
    return data

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

@pytest.mark.anyio
async def test_get_data():
    # async with AsyncClient(base_url='http://127.0.0.1:8000') as ac: # don't use `app=app`, need manual launch server, but it's ok.
    async with AsyncClient(app=app, base_url='http://127.0.0.1:8000') as ac:
        response = await ac.get('/')
    assert response.status_code == 200

I cannot reproduce this anymore... 🤔 It works fine here.

[ia-bg]

@Kludex because you use the

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

in your code ;)

The documentation still needs an update.

[CarlosLannister]

I am receiving the same error if I extend @Kludex code and add some value in ac.get

@pytest.mark.anyio
async def test_get_data():
    # async with AsyncClient(base_url='http://127.0.0.1:8000') as ac: # don't use `app=app`, need manual launch server, but it's ok.
    async with AsyncClient(app=app, base_url='http://127.0.0.1:8000') as ac:
        response = await ac.get('/items/foo')
    assert response.status_code == 200

[CarlosLannister]

If I manually launch the server seems to be working

@pytest.mark.anyio
async def test_get_data():
    async with AsyncClient(base_url='http://127.0.0.1:8000') as ac: # don't use `app=app`, need manual launch server, but it's ok.
        response = await ac.get('/items/foo')
    assert response.status_code == 200

[MartinSchere]

@CarlosLannister The base URL needs to be http://test:

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