Async Tests¶

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

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

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

pytest.mark.anyio¶

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

HTTPX¶

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

TestClient 在内部做了一些魔法，以便在标准的 pytest 中，在普通的 def 测试函数中调用异步 FastAPI 应用程序。但当我们开始在异步函数中使用它时，那种魔法就不再奏效了。通过异步运行我们的测试，我们不能再在测试函数中使用 TestClient 了。

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

示例¶

对于一个简单的例子，让我们考虑一个类似于 Bigger Applications 和 Testing 中描述的文件结构

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

main.py 文件将包含

Python 3.9+

from fastapi import FastAPI

app = FastAPI()


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

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

Python 3.9+

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%

In Detail¶

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

Python 3.9+

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

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 - 请求是异步的。

警告

如果您的应用程序依赖于 lifespan 事件，AsyncClient 将不会触发这些事件。为了确保它们被触发，请使用来自 florimondmanca/asgi-lifespan 的 LifespanManager。

Other Asynchronous Function Calls¶

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

提示

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