使用 pytest-asyncio 测试 FastAPI 路由时出现“RuntimeError：事件循环已关闭”

我收到错误

运行时错误：事件循环已关闭

每次我尝试在测试中进行多个异步调用时。我已经尝试使用其他 Stack Overflow 帖子中的所有其他建议来重写event_loop固定装置但没有任何作用。我想知道我错过了什么？

运行测试命令：

python -m pytest tests/ --asyncio-mode=auto

要求.txt

pytest==7.1.2
pytest-asyncio==0.18.3
pytest-html==3.1.1
pytest-metadata==2.0.1

test.py

async def test_user(test_client_fast_api):
    assert 200 == 200

    # works fine
    request_first = test_client_fast_api.post("/first_route")

    # recieve RuntimeError: Event loop is closed
    request_second = test_client_fast_api.post("/second_route")

测试.py

@pytest.fixture()
def event_loop():
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
    yield loop
    loop.close()

添加文件conftest.py到放置测试脚本的目录。

并编写以下代码：

import pytest
from main import app
from httpx import AsyncClient

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture(scope="session")
async def client():
    async with AsyncClient(app=app, base_url="http://test") as client:
        print("Client is ready")
        yield client

然后在您自己的测试代码中使用这些装置。例如，这是我自己项目的真实测试代码。您可以将其更改为您自己的。

import pytest
from httpx import AsyncClient

@pytest.mark.anyio
async def test_run_not_exists_schedule(client: AsyncClient):
    response = await client.get("/schedule/list")
    assert response.status_code == 200
    schedules = response.json()["data"]["schedules"]
    schedules_exists = [i["id"] for i in schedules]
    not_exists_id = max(schedules_exists) + 1
    request_body = {"id": not_exists_id}
    response = await client.put("/schedule/run_cycle", data=request_body)
    assert response.status_code != 200  

@pytest.mark.anyio
async def test_run_adfasdfw(client: AsyncClient):
    response = await client.get("/schedule/list")
    assert response.status_code == 200
    schedules = response.json()["data"]["schedules"]
    schedules_exists = [i["id"] for i in schedules]
    not_exists_id = max(schedules_exists) + 1
    request_body = {"id": not_exists_id}
    response = await client.put("/schedule/run_cycle", data=request_body)
    assert response.status_code != 200

最后在项目的终端中运行

python -m pytest

如果一切顺利的话应该没问题。

这可能涉及到需要安装的库。

pytest
httpx