非同步測試¶

🌐 AI 與人類共同完成的翻譯

此翻譯由人類指導的 AI 完成。🤝

可能會有對原意的誤解，或讀起來不自然等問題。🤖

你可以透過協助我們更好地引導 AI LLM來改進此翻譯。

英文版

你已經看過如何使用提供的 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 會是：

Python 3.10+

from fastapi import FastAPI

app = FastAPI()


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

檔案 test_main.py 會包含針對 main.py 的測試，現在可能像這樣：

Python 3.10+

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

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 發送非同步請求。

Python 3.10+

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 需搭配 async/await —— 請求是非同步的。

Warning

如果你的應用仰賴 lifespan 事件，AsyncClient 不會觸發這些事件。若要確保它們被觸發，請使用 florimondmanca/asgi-lifespan 的 LifespanManager。

其他非同步函式呼叫¶

由於測試函式現在是非同步的，你也可以在測試中呼叫（並 await）其他 async 函式，和在程式碼其他地方一樣。

Tip

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