我遇到过这个问题，虽然它一定是常见的，但我看不到任何解决方案。所以，也许我错过了什么。
我正在开发具有异步端点和数据库异步连接的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_venvlibsite-packagesfastapirouting.py:217: in app
solved_result = await solve_dependencies(
coin_venvlibsite-packagesfastapidependenciesutils.py:529: in solve_dependencies
solved = await run_in_threadpool(call, **sub_values)
AttributeError: module 'anyio' has no attribute 'to_thread'
我得出的结论是，只有当端点中存在依赖项时，才会出现错误。奇怪的是，我的环境中甚至没有anyio。
那么，有没有一种方法可以测试具有依赖关系和异步数据库连接的异步FastAPI端点？当然，肯定有什么，这种情况并不是独一无二的。。。
UPD:我尝试使用decorator@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来测试异步代码。不管怎样，我不需要第二次测试，为什么它在这里，我该如何摆脱它？
事实证明，我可以指定后端来运行这样的测试：
@pytest.fixture
def anyio_backend():
return 'asyncio'
所以，现在我只有正确的测试运行(
pytest在不同的事件循环(而不是get_running_loop(上运行，因此，当您尝试在同一上下文中运行它时，它会引发异常。我建议您考虑使用nestrongyncio(https://pypi.org/project/nest-asyncio/)，以便pytest可以在同一事件循环中运行。
import nest_asyncio
nest_asyncio.apply()