异步测试¶

你已经了解了如何使用提供的 TestClient 测试 FastAPI 应用程序。到目前为止，你只看到了如何编写同步测试，而没有使用 async 函数。

在测试中使用异步函数可能很有用，例如，当你需要异步查询数据库时。假设你想要测试向 FastAPI 应用程序发送请求，然后验证后端在使用异步数据库库时是否正确地将数据写入了数据库。

让我们看看如何实现这一点。

pytest.mark.anyio¶

如果我们想在测试中调用异步函数，我们的测试函数本身必须是异步的。AnyIO 为此提供了一个很好的插件，它允许我们指定某些测试函数应被异步调用。

HTTPX¶

即使你的 FastAPI 应用程序使用普通的 def 函数而不是 async def，它在底层仍然是一个异步应用程序。

TestClient 内部做了一些处理，使其能够在普通的 def 测试函数中使用标准的 pytest 来调用异步的 FastAPI 应用程序。但是，当我们在异步函数内部使用它时，这种处理就不再有效。通过异步运行测试，我们无法再在测试函数内部使用 TestClient。

TestClient 基于 HTTPX，幸运的是，我们可以直接使用它来测试 API。

示例¶

对于一个简单的示例，我们考虑一个类似于大型应用程序和测试中描述的文件结构：

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

文件 main.py 的内容如下：

Python 3.8+

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Tomato"}

文件 test_main.py 包含 main.py 的测试，现在它可能看起来像这样：

Python 3.8+

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

运行它¶

你可以像往常一样通过以下命令运行测试：

$ pytest

---> 100%

详细说明¶

标记 @pytest.mark.anyio 告诉 pytest 这个测试函数应该被异步调用：

Python 3.8+

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

Tip

请注意，测试函数现在是 async def，而不像之前使用 TestClient 时只是 def。

然后，我们可以使用应用程序创建一个 AsyncClient，并使用 await 向其发送异步请求。

Python 3.8+

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

这等同于我们之前使用 TestClient 发出请求的代码：

response = client.get('/')

Tip

请注意，我们在使用新的 AsyncClient 时使用了 async/await - 请求是异步的。

Warning

如果你的应用程序依赖于生命周期事件，AsyncClient 将不会触发这些事件。要确保它们被触发，请使用 florimondmanca/asgi-lifespan 中的 LifespanManager。

其他异步函数调用¶

由于测试函数现在是异步的，你现在除了可以在测试中向 FastAPI 应用程序发送请求外，还可以调用（并 await）其他 async 函数，就像你在代码中任何其他地方调用它们一样。

Tip

如果在测试中集成异步函数调用时遇到 RuntimeError: Task attached to a different loop（例如，在使用 MongoDB 的 MotorClient 时），请记住，需要事件循环的对象实例化应仅在异步函数内进行，例如，在 '@app.on_event("startup") 回调中。