Testing asynchronous FastAPI endpoints with dependencies

Question

I've encountered this problem, and I can't see any solution, though it must be a common one. So, maybe I'm missing something here.

I'm working on FastAPI app with asynchronous endpoints and asynchronous connection with database. Database connection is passed as a dependency. I want to write some asynchronous tests for said app.

engine = create_async_engine(connection_string, echo=True)

def get_session():
    return sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@router.post("/register")
async def register(
    user_data: UserRequest,
    authorize: AuthJWT = Depends(),
    async_session: sessionmaker = Depends(get_session),
):
    """Register new user."""
    if authorize.get_jwt_subject():
        raise LogicException("already authorized")

    session: AsyncSession
    async with async_session() as session:
        query = await session.execute(
            select(UserModel).where(UserModel.name == user_data.name)
        )
    ...

I'm using AsyncSession to work with database. So in my test, db connection also has to be asynchronous.

engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
app.dependency_overrides[get_session] = lambda: sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

@pytest.mark.asyncio
async def test_create_user():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post(
            "/register",
            json={"name": "TestGuy", "password": "TestPass"},
        )
        assert response.status_code == 200, response.text

When running the test, I get the following error:

...
coin_venv\lib\site-packages\fastapi\routing.py:217: in app
    solved_result = await solve_dependencies(
coin_venv\lib\site-packages\fastapi\dependencies\utils.py:529: in solve_dependencies
    solved = await run_in_threadpool(call, **sub_values)
AttributeError: module 'anyio' has no attribute 'to_thread'

I concluded that error appears only when there is a dependency in an endpoint. Weird part is that I don't even have anyio in my environment.

So, is there a way to test asynchronous FastAPI endpoints with dependencies and asynchronous db connection? Surely, there must be something, it's not like this situation is something unique...

UPD: I tried using decorator @pytest.mark.anyio and also have installed trio and anyio. Now pytest seem to discover two distinct tests in this one:

login_test.py::test_create_user[asyncio]
login_test.py::test_create_user[trio]

Both fails, first one with what seems to be a valid error in my code, and second one with:

RuntimeError: There is no current event loop in thread 'MainThread'.

I guess it is true, though I don't really know if pytest creates eventloop to test async code. Anyway, I don't need the second test, why it is here, and how can I get rid of it?

user_6479666 · Answer

It turned out, I can specify backend to run tests like this:

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

So, now I have only the right tests running)

user_13503720 · Answer

pytest runs on different eventloop (not get_running_loop), and so, when you try to run it in the same context, it raises exception. I suggest you to consider using nest_asyncio (https://pypi.org/project/nest-asyncio/), so that pytest can run in the same eventloop.

import nest_asyncio
nest_asyncio.apply()

Testing asynchronous FastAPI endpoints with dependencies

2 Answers