跳至內容

測試

感謝 Starlette,測試 FastAPI 應用程式變得輕鬆又愉快。

它基於 HTTPX,而 HTTPX 又是基於 Requests 設計的,因此它非常熟悉且直觀。

使用它,您可以直接將 pytestFastAPI 搭配使用。

使用 TestClient

資訊

要使用 TestClient,請先安裝 httpx

請確保您建立一個 虛擬環境,啟動它,然後安裝它,例如:

$ pip install httpx

匯入 TestClient

透過將您的 FastAPI 應用程式傳遞給它來建立 TestClient

建立名稱以 test_ 開頭的函式(這是標準的 pytest 慣例)。

以與使用 httpx 相同的方式使用 TestClient 物件。

使用您需要檢查的標準 Python 運算式編寫簡單的 assert 陳述式(同樣是標準的 pytest)。

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}


client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

提示

請注意,測試函式是普通的 def,而不是 async def

而且對 client 的呼叫也是普通的呼叫,而不是使用 await

這讓您可以直接使用 pytest 而不產生複雜性。

「技術細節」

您也可以使用 from starlette.testclient import TestClient

FastAPI 提供與 starlette.testclient 相同的 fastapi.testclient,只是為了方便您,開發人員。但它直接來自 Starlette。

提示

如果您想在測試中呼叫 async 函式,除了向您的 FastAPI 應用程式發送請求之外(例如非同步資料庫函式),請查看進階教學中的 非同步測試

分離測試

在實際應用程式中,您可能會將測試放在不同的檔案中。

而且您的 FastAPI 應用程式也可能由多個檔案/模組等組成。

FastAPI 應用程式檔案

假設您有一個如 更大的應用程式 中所述的檔案結構

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

main.py 檔案中,您有您的 FastAPI 應用程式

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

測試檔案

然後您可以有一個包含測試的 test_main.py 檔案。它可以位於同一個 Python 套件中(同一個目錄中有一個 __init__.py 檔案)

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

因為此檔案位於同一個套件中,您可以使用相對匯入從 main 模組 (main.py) 匯入物件 app

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

...然後像之前一樣編寫測試程式碼。

測試:擴展範例

現在讓我們擴展此範例並添加更多細節,以了解如何測試不同的部分。

擴展的 FastAPI 應用程式檔案

讓我們繼續使用與之前相同的檔案結構

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

假設現在包含 FastAPI 應用程式的 main.py 檔案有一些其他的**路徑操作**。

它有一個可能會返回錯誤的 GET 操作。

它有一個 POST 操作可能會回傳多種錯誤。

兩個路徑操作都需要一個 X-Token 標頭。

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item
from typing import Annotated, Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item
from typing import Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from typing_extensions import Annotated

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

提示

如果可能,建議使用 Annotated 版本。

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

提示

如果可能,建議使用 Annotated 版本。

from typing import Union

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: Union[str, None] = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

擴展測試檔案

然後您可以用擴展的測試更新 test_main.py

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_item():
    response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 200
    assert response.json() == {
        "id": "foo",
        "title": "Foo",
        "description": "There goes my hero",
    }


def test_read_item_bad_token():
    response = client.get("/items/foo", headers={"X-Token": "hailhydra"})
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_read_nonexistent_item():
    response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}


def test_create_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
    )
    assert response.status_code == 200
    assert response.json() == {
        "id": "foobar",
        "title": "Foo Bar",
        "description": "The Foo Barters",
    }


def test_create_item_bad_token():
    response = client.post(
        "/items/",
        headers={"X-Token": "hailhydra"},
        json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},
    )
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_create_existing_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={
            "id": "foo",
            "title": "The Foo ID Stealers",
            "description": "There goes my stealer",
        },
    )
    assert response.status_code == 409
    assert response.json() == {"detail": "Item already exists"}

當您需要客戶端在請求中傳遞資訊,但不知道該怎麼做時,您可以搜尋(Google)如何在 httpx 中執行此操作,或者甚至是如何在 requests 中執行此操作,因為 HTTPX 的設計是基於 Requests 的設計。

然後您只需在測試中執行相同的操作即可。

例如:

  • 要傳遞路徑查詢參數,請將其添加到 URL 本身。
  • 要傳遞 JSON 主體,請將 Python 物件(例如 dict)傳遞給參數 json
  • 如果您需要發送表單資料而不是 JSON,請改用 data 參數。
  • 要傳遞標頭,請在 headers 參數中使用 dict
  • 對於Cookie,請在 cookies 參數中使用 dict

有關如何將資料傳遞到後端(使用 httpxTestClient)的更多資訊,請查看 HTTPX 文件

資訊

請注意,TestClient 接收的是可以轉換為 JSON 的資料,而不是 Pydantic 模型。

如果您的測試中有一個 Pydantic 模型,並且您想在測試期間將其資料發送到應用程式,您可以使用 JSON 相容編碼器 中描述的 jsonable_encoder

執行它

之後,您只需要安裝 pytest

請確保您建立一個 虛擬環境,啟動它,然後安裝它,例如:

$ pip install pytest

---> 100%

它會自動偵測檔案和測試,執行它們,並將結果回報給您。

使用以下指令執行測試

$ pytest

================ test session starts ================
platform linux -- Python 3.6.9, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: /home/user/code/superawesome-cli/app
plugins: forked-1.1.3, xdist-1.31.0, cov-2.8.1
collected 6 items

---> 100%

test_main.py <span style="color: green; white-space: pre;">......                            [100%]</span>

<span style="color: green;">================= 1 passed in 0.03s =================</span>