异步测试¶

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

在测试中能够使用异步函数是非常有用的，例如，当你异步查询数据库时。想象一下，你想要测试向 FastAPI 应用程序发送请求，然后在使用异步数据库库的同时，验证后端是否成功将正确的数据写入数据库。

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

pytest.mark.anyio¶

如果我们想在测试中调用异步函数，我们的测试函数必须是异步的。AnyIO 提供了一个非常棒的插件，允许我们指定某些测试函数需要以异步方式调用。

HTTPX¶

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

TestClient 在内部通过一些魔法手段，在你的普通 def 测试函数中使用标准 pytest 调用异步 FastAPI 应用程序。但是，当我们在异步函数中使用它时，这种魔法就不再起作用了。通过异步运行测试，我们不能再在测试函数中使用 TestClient。

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

示例¶

作为一个简单的示例，让我们考虑一个类似于大型应用和测试中描述的文件结构。

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

main.py 文件内容如下

Python 3.10+

from fastapi import FastAPI

app = FastAPI()


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

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

Python 3.10+

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.10+

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"}

提示

注意，现在测试函数是 async def，而不是之前使用 TestClient 时的普通 def。

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

Python 3.10+

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"}

这等同于

response = client.get('/')

...我们过去用来使用 TestClient 发送请求的方式。

提示

注意，我们正在将 async/await 与新的 AsyncClient 一起使用 - 请求是异步的。

警告

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

其他异步函数调用¶

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

提示

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