异步测试¶

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

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

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

pytest.mark.anyio¶

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

HTTPX¶

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

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

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

示例¶

为了简单起见，我们考虑一个类似于 Bigger Applications 和 Testing 中描述的文件结构：

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

main.py 文件将包含：

from fastapi import FastAPI

app = FastAPI()


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

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

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 这个测试函数应该是异步调用的：

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 向其发送异步请求。

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 进行请求的方式。

Tip

注意，我们正在使用新的 AsyncClient 进行异步/等待 - 请求是异步的。

Warning

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

其他异步函数调用¶

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

Tip

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