异步测试¶

您已经了解了如何使用 **FastAPI** 提供的 `TestClient` 来测试您的应用程序。到目前为止，您只看到了如何编写同步测试，而没有使用 `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.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"}

提示

请注意，现在测试函数是 `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"}

这相当于

response = client.get('/')

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

提示

请注意，我们正在使用新的 `AsyncClient` 进行 async/await 操作——请求是异步的。

警告

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

其他异步函数调用¶

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

提示

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