python - 使用依赖项测试异步 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 是否创建事件循环来测试异步代码。无论如何，我不需要第二个测试，为什么它会在这里，我该如何摆脱它？

Answer 1

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

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

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