使用pytest-asyncio测试FastAPI路由时出现了"RuntimeError: Event loop is closed"错误。

Question

我在进行测试时，每当尝试进行多个异步调用时都会收到以下错误：

RuntimeError: Event loop is closed

我已经尝试了其他Stack Overflow帖子中的所有建议来重写event_loop fixture，但没有任何作用。我想知道我错过了什么？

测试命令：

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

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

conftest.py

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

- Dmitriy_kzn

Answer 1

将文件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

- Bai Jinge

---

如果这个答案不够用，请尝试包含另一个答案：https://dev59.com/FMTsa4cB1Zd3GeqPHtK4#73019163。两个答案的组合对我很有用。 - Gino Mempin

Answer 2

这是我使用的事件循环装置和TestClient模式：

from asyncio import get_event_loop
from unittest import TestCase

from async_asgi_testclient import TestClient

@pytest.fixture(scope="module")
def event_loop():
    loop = get_event_loop()
    yield loop

@pytest.mark.asyncio
    async def test_example_test_case(self):
        async with TestClient(app) as async_client:
            response = await async_client.get(
                "/api/v1/example",
                query_string={"example": "param"},
            )
        assert response.status_code == HTTP_200_OK

请参考相关的 GitHub 问题: https://github.com/tiangolo/fastapi/issues/2006#issuecomment-689611040

---

请注意 - 我无法弄清如何使用基于类的测试。 无论是`unittest.TestCase`还是`asynctest.case.TestCase` 都不能为我工作。`pytest-asyncio`文档(here)指出：

不支持基于标准unittest库的测试类继承，建议用户使用unittest.IsolatedAsyncioTestCase或像asynctest这样的异步框架。

- wjkw1