FastAPIアプリケーションで行う様々なテストパターンのまとめ
2024.05.05はじめに
データアナリティクス事業本部のkobayashiです。
FastAPIで作成したエンドポイントに対するPytestでテストコードはリクエストの送り方で書き方が変わりますのでそのパターンをまとめたいと思います。
FastAPI TestClientを使ってテストコードを作成する
FastAPIではテスト用のクラスであるTestClientが用意されていてこれを使用することでrequestsパッケージなどを使用してリクエストを送ってテスト作成するのではなく直接FastAPIのメソッドを実行してFastAPIのアプリケーションをテストできます。
環境
- Python: 0.111.0
TestClientのメソッド
TestClientにはさまざまなメソッドが用意されています。FastAPIでエンドポイントを作成する場合には@app.get()デコレータなどでRESTエンドポイントを作成します。したがって、例えばテスト対象がRESTのリクエストメソッドgetであればTestClientで使うのもこれに合わせたgetメソッドとなります。
では早速そのパターンをまとめたいと思います。
単純なリクエストパターン
パラメータが不要なエンドポイントのテスト
以下のようなGETのリクエストメソッドに対するテストを記述します。
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_main():
return {"msg": "Hello World"}
GETなのでTestClientのメソッドもgetメソッドを使って記述します。
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"}
パスパラメータを使ったリクエストのテスト
GETのリクエストメソッドに対するテストを記述しますが先程と違いパスパラメータを引数として取るエンドポイントに対するテストを記述します。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id):
return {"item_id": item_id}
先ほどと同じGETなのでTestClientのメソッドもgetメソッドを使って記述しますが、パスパラメータの箇所が違うのでここにテストしたい値を入れます。
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["item_id"],
[
pytest.param(
"hogehoge",
),
],
)
def test_read_main(item_id: str):
response = client.get(f"/items/{item_id}")
assert response.status_code == 200
assert response.json() == {"item_id": item_id}
クエリパラメータを使ったリクエストのテスト
次はクエリパタメータを使ったエンドポイントに対するテストを記述します。
from fastapi import FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
return fake_items_db[skip : skip + limit]
クエリパタメータを使ったエンドポイントに対するリクエストにはTestClientの各メソッドのparamsパラメータを使います。
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["skip","limit"],
[
pytest.param(
0,2
),
],
)
def test_read_item(skip: int, limit: int):
response = client.get(
"/items/",
params={"skip": skip, "limit": limit},
)
assert response.status_code == 200
assert response.json() == [{'item_name': 'Foo'}, {'item_name': 'Bar'}]
一応paramsパラメータを使わず原始的に以下のような記述も可能ですが、paramsパラメータを使ったほうが拡張性は高いのであまりこの記述は使いません。
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["skip","limit"],
[
pytest.param(
0,2
),
],
)
def test_read_item(skip: int, limit: int):
response = client.get(
f"/items/?skip={skip}&limit={limit}",
)
assert response.status_code == 200
assert response.json() == [{'item_name': 'Foo'}, {'item_name': 'Bar'}]
リクエストボディを使ったリクエストのテスト
POSTのリクエストメソッドでリクエストボディを使ったエンドポイントに対するテストを記述します。
from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
app = FastAPI()
@app.post("/items/")
async def create_item(item: Item):
return item
POSTなのでTestClientのメソッドもpostメソッドを使って記述し、リクエストボディを使ったエンドポイントに対するリクエストにはTestClientの各メソッドのjsonパラメータを使います。
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["item"],
[
pytest.param(
{
"name": "Foo",
"description": "An optional description",
"price": 45.2,
"tax": 3.5
}
),
],
)
def test_create_item(item: dict):
response = client.post(
"/items/",
json=item,
)
assert response.status_code == 200
assert response.json() == item
ヘッダーのパラメータを使ったリクエストのテスト
ヘッダーのパラメータを使ったリクエストに対するテストを記述します。このエンドポイントの処理はリクエストヘッダーに含まれるuser_agentの値をそのまま返す処理になります。
from typing import Union
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(user_agent: Union[str, None] = Header(default=None)):
return {"User-Agent": user_agent}
リクエストヘッダーを使ったエンドポイントに対するリクエストにはTestClientの各メソッドのheadersパラメータを使います
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["user_agent"],
[
pytest.param(
"user_agent_sample"
),
],
)
def test_read_items(user_agent: str):
response = client.get(
"/items/",
headers={"user_agent": user_agent},
)
assert response.status_code == 200
assert response.json() == {"User-Agent": user_agent}
クッキーのパラメータを使ったリクエストのテスト
クッキーのパラメータを使ったリクエストに対するテストを記述します。このエンドポイントの処理はcookieにads_idがある場合にその中身を返すエンドポイントになります。
from typing import Union
from fastapi import Cookie, FastAPI
app = FastAPI()
@app.get("/items/")
async def read_items(ads_id: Union[str, None] = Cookie(default=None)):
return {"ads_id": ads_id}
cookieもHTTPリクエストヘッダーの1種類ですがheadersパラメータではなくTestClientの各メソッドのcookiesパラメータを使います。
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["ads_id"],
[
pytest.param(
"ads_id_sample"
),
],
)
def test_read_items(ads_id: str):
response = client.get(
"/items/",
cookies={"ads_id": ads_id},
)
assert response.status_code == 200
assert response.json() == {"ads_id": ads_id}
フォームデータを使ったリクエストのテスト
フォームデータを使ったリクエストに対するテスを記述します。
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login/")
def login(username: str = Form(), password: str = Form()):
return {"username": username}
フォームデータを使ったエンドポイントに対するリクエストにはTestClientの各メソッドのdataパラメータを使います。dataパラメータはdict形式になりますが、キーはエンドポイントのメソッドでFormクラスで指定した引数名と合わせる必要があります。これを間違えると422 Unprocessable Entityのエラーが出ます。
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["username","password"],
[
pytest.param(
"username_1", "password_1"
),
],
)
def test_login(username: str, password: str):
response = client.post(
"/login/",
data={"username": username, "password": password},
)
assert response.status_code == 200
assert response.json() == {"username": username}
ファイルアップロードを行うリクエストのテスト
FastAPIでファイルを送信するリクエストに対するテストを記述します。
from typing import Annotated
from fastapi import FastAPI, File
app = FastAPI()
@app.post("/files/")
def create_file(file: Annotated[bytes, File()]):
return {"file_size": len(file)}
ファイル送信を行うエンドポイントに対するリクエストにはTestClientの各メソッドのfilesパラメータを使います。filesパラメータはdict形式になりますが、キーはエンドポイントのメソッドでFileクラスで指定した引数名と合わせる必要があります。これを間違えると先ほどと同じ様に422 Unprocessable Entityのエラーが出ます。
import os
import tempfile
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["file_name", "file_content"],
[
pytest.param(
"file_name_1", "sample text 0123456789"
),
],
)
def test_login(file_name: str, file_content: str):
with tempfile.TemporaryDirectory(dir="/tmp/") as tmpdir:
filename = os.path.join(tmpdir, file_name)
with open(filename, "w+") as f:
f.write(file_content)
file_data = {"file": (file_name, open(filename, "rb"))}
response = client.post(
"/files/",
files=file_data,
)
assert response.status_code == 200
assert response.json() == {"file_size": len(file_content)}
複合リクエストパターン
先ほどまではそれぞれシンプルなリクエストパターンでしたが、FastAPIを使ってエンドポイントを実装していくといくつかのリクエストを組み合わせる処置が必要になってきます。次はこのパターンを考えてみます。
CSRFトークンで保護されたエンドポイントへのテスト
POSTのリクエストメソッドを行うエンドポイントがありますが、CSRF対策でCSRFトークンが必要なエンドポイントに対するテストを記述してみます。
このエンドポイントの処理の想定としてははじめに/csrf_token/エンドポイントにGETリクエストメソッドでリクエストを送りレスポンスとして{"csrf_token":{CSRFトークン}"} が返ってきます。このCSRFトークンを使って/items/エンドポイントにPOSTリクエストメソッドでitemを追加します。
このエンドポイントに対するテストは以下のようになります。
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["item"],
[
pytest.param(
{
"name": "Foo",
"description": "An optional description",
"price": 45.2,
"tax": 3.5
}
),
],
)
def test_create_item(item: dict):
# CSRFトークンを取得するTestClientの処理
response_csrf = client.get("/csrf_token/")
response_csrf_body = response_csrf.json()
csrf_token = response_csrf_body["csrf_token"]
# itemを追加する処理
response = client.post(
"/items/",
json=item,
headers={"x-csrf-token": csrf_token}
)
assert response.status_code == 200
assert response.json() == item
ポイントとしては先にCSRFトークンをTestClientのgetメソッドで取得し、その結果を使ってitemを追加するPOSTメソッドのテストを行っています。
Loginが必要なエンドポイントへのテスト
Login処理が必要なエンドポイントに対するテストを記述してみます。
このエンドポイントの処理の想定としてははじめに/login/エンドポイントにPOSTリクエストメソッドでリクエストを送りレスポンスとして{"access_token":{Accessトークン}"} が返ってきます。このAccessトークンを使って/items/エンドポイントにPOSTリクエストメソッドでitemを追加します。さらに先ほどと同じ様にPOSTはCSRF対策がされているとします。
このエンドポイントに対するテストは以下のようになります。
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
@pytest.mark.parametrize(
["username", "password", "item"],
[
pytest.param(
"username_1", "password_1"
{
"name": "Foo",
"description": "An optional description",
"price": 45.2,
"tax": 3.5
}
),
],
)
def test_create_item(username: str, password: str, item: dict):
# CSRFトークンを取得するTestClientの処理
response_csrf = client.get("/csrf_token/")
response_csrf_body = response_csrf.json()
csrf_token = response_csrf_body["csrf_token"]
# Accessトークンを取得するTestClientの処理
response = client.post(
"/login/",
data={"username": username, "password": password},
headers={"x-csrf-token": csrf_token},
)
response_csrf_body = response_csrf.json()
access_token = response_csrf_body["access_token"]
# CSRFトークンを取得するTestClientの処理(2度目）
response_csrf = client.get("/csrf_token/")
response_csrf_body = response_csrf.json()
csrf_token = response_csrf_body["csrf_token"]
# itemを追加する処理
response = client.post(
"/items/",
json=item,
cookies={"access_token": access_token},
headers={"x-csrf-token": csrf_token}
)
assert response.status_code == 200
assert response.json() == item
ポイントしてしてははじめにCSRFトークンを取得しその結果を使ってLogin処理をします。Login処理ではAccessトークンが返ってくるのでAccessトークンを使うのと再びCSRFトークンを取得してからitemの追加処理のテストをしています。
まとめ
FastAPIで作成したエンドポイントに対するPytestでテストコードの様々なパターンを記述しました。組み合わせパターンで記述したCSRFトークンの取得やログイン処理はより高度なテストを記述するためにこれらをfixtureとして共通化することがポイントかと思います。どなたかの参考になれば幸いです。
最後まで読んで頂いてありがとうございました。
