非同步測試¶

您已經看過如何使用提供的 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 將包含

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。

其他非同步函式呼叫¶

由於測試函式現在是非同步的，除了在測試中向 FastAPI 應用程式發送請求之外，您現在還可以呼叫（和 await）其他 async 函式，就像在程式碼中的其他任何地方呼叫它們一樣。

提示

如果您在測試中整合非同步函式呼叫時遇到 RuntimeError: Task attached to a different loop（例如，使用 MongoDB 的 MotorClient 時），請記住僅在非同步函式中實例化需要事件迴圈的物件，例如 '@app.on_event("startup") 回呼。