测试具有依赖项的异步 FastAPI 端点

Question

我遇到过这个问题，虽然它一定是一个常见问题，但我看不到任何解决方案。所以，也许我在这里遗漏了一些东西。

我正在开发具有异步端点和与数据库异步连接的 FastAPI 应用程序。数据库连接作为依赖项传递。我想为所述应用程序编写一些异步测试。

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

我正在使用 AsyncSession 来处理数据库。所以在我的测试中，数据库连接也必须是异步的。

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

运行测试时，我收到以下错误：

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

我得出的结论是，仅当端点存在依赖性时才会出现错误。奇怪的是我的环境中甚至没有anyio。

那么，有没有一种方法可以测试具有依赖项和异步数据库连接的异步 FastAPI 端点？当然，一定有什么，这种情况并不是什么独特的事情……

UPD：我尝试使用装饰器@pytest.mark.anyio，并且还安装了trio和anyio。现在 pytest 似乎在其中发现了两个不同的测试：

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

两者都失败了，第一个在我的代码中似乎是一个有效的错误，第二个是：

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

我想这是真的，尽管我真的不知道 pytest 是否创建 eventloop 来测试异步代码。无论如何，我不需要第二次测试，为什么它会在这里，我该如何摆脱它？

Answer 1

事实证明，我可以指定后端来运行这样的测试：

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

所以，现在我只运行正确的测试）