异步测试¶

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

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

让我们看看如何使其工作。

pytest.mark.anyio¶

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

HTTPX¶

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

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

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

示例¶

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

.
├── 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"}

提示

请注意，测试函数现在是 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 发出请求。

提示

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

警告

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

其他异步函数调用¶

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

提示

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