はじめに
株式会社YUZURIHAで2024年6月からインターンをさせていただいている大学生です。
インターンを終えるにあたり、学んだことをアウトプットする機会として記事を作成させていただけることになったので、インターン中で最も苦労した部分である、テストについて自分なりのTipsをまとめてみました。
概要
FastAPIは非常に高性能で柔軟なフレームワークですが、そのテストの書き方には工夫が必要です。本記事では、非同期処理を含むFastAPIのテスト方法について、自分が実際に書いたコードを紹介したいと思います。その中で、pytestを使ったユニットテストとAPIテストの書き方、そしてテストを自動化するためのCIの設定について詳しく自分の得た知見を共有したいと思います。具体的には、以下のポイントに重点を置いています。
-
非同期テストの実践
FastAPIの非同期テストの書き方と、それに必要な設定を紹介します。 -
pytestの基本と活用方法
pytestのfixtureやデコレータの使い方、parametrizeを使った効率的なテスト方法を紹介します。 -
CIの実装
GitHub Actionsを使ったCIの設定方法について、Dockerを用いた開発環境下でのユニットテストとAPIテストの実行方法ついて自分の書いたコードの説明を行います。
この記事を書こうと思った理由
最近インターン等でテストコードを書く機会が増えてきて、DBが絡んでくるテストやFastAPI特有のテスト手法等があり、とても苦労しましたが試行錯誤の末に自分なりの効果的なテストの書き方を見つけました。この知見を共有することで、同じような課題を抱える開発者の助けになりたいと考え、本記事を書きました。
記事を読む上での注意点
筆者自身そこまでテストコードに精通しているわけではないため、一部おかしな記述や不正確な情報が含まれているかもしれません。
もしお気付きの点やご指摘があれば、コメントで教えていただけると幸いです。
この記事の対象読者
- FastAPIのテストを書いてみたい
- pytestを用いてテストを行いたい
- SQLite等を用いたテスト環境ならではの環境ではなく、本番環境に近い環境でテストを行いたい
FastAPIのテストTips
非同期テスト
FastAPIの大きな特徴として、ASGIであることが挙げられます。そのおかげで非同期処理による高速な動作を実現しています。この特徴を用いた非同期関数のテストを行う際には、デコレータで@pytest.mark.asyncioを付与する必要があります。公式ドキュメントの方には@pytest.mark.anyioと書かれているのですが、どちらで実行しても正しく動作します。
補足
ASGI (Asynchronous Server Gateway Interface) は、Pythonの非同期Webフレームワークとサーバーの間の標準インターフェースです。ASGIは非同期処理をサポートし、WebSocketなどのプロトコルにも対応できるため、より高速でリアルタイムなアプリケーションの開発が可能です。
WSGI (Web Server Gateway Interface) は、Pythonの同期Webフレームワークとサーバーの間の標準インターフェースです。WSGIはシンプルで広く採用されていますが、同期処理のみをサポートしているため、リクエストが多いときにはスケーラビリティが制限されることがあります。
FastAPIはASGIを使用しているため、非同期処理を活用した高速なパフォーマンスを実現できます。
pytestの便利な機能
fixtureでテストの前後処理を記述する
pytestではfixtureを定義することでテストの前後の処理を定義することができます。例えば、テストの前にAPIテスト用のクライアントを用意したり、テストで用いるdbのセッションを作成し、テストが終了したらdbのをロールバックしたり、dbセッションをクローズするなどの処理を行うことができます。
@pytest.fixture(scope="function")
def test_setup():
# テストの前処理
db = SessionLocal()
yield db # ここでテストが実行される
# テストの後処理
db.rollback()
db.close()
またこのfixtureはconftest.pyというファイルに記述することで複数定義することができます。例えば以下のようなファイル構造をとると、統合テストを行う際には(1)と(2)のconftest.pyのfixtureが適用され、単体テストを行う際には(1)と(3)のconftest.pyのfixtureが適用されます。このようにしてfixtureの共通化とテストの種類ごとにfixtureの個別化を行うことが可能です。
── test
├── conftest.py ・・・（1）
├── integration_test
│ ├── conftest.py ・・・（2）
│ ├── test_auth_api.py
| ...
└── unit_test
├── conftest.py ・・・（3）
├── test_auth.py
...
parametrizeでテストを共通化する
テストを記述する際に、さまざまなパラメータを関数の引数に与えてテストを行うことはしばしばありますが、その際テストコードの大部分が同じで、assertの部分だけが異なるテスト関数が大量にできてしまうことがあると思います。
pytestではテスト関数に与えるパラメータを複数用意して、1つのテスト関数の中で複数の状況下のテストを行うことが可能です。
@pytest.mark.parametrize(
["test_input", "expected"],
[
pytest.param("test_input1", "expected_output1"),
pytest.param("test_input2", "expected_output2"),
pytest.param("test_input3", "expected_output3"),
]
)
def test_with_parametrize(test_input, expected):
test_param = test_input
actual = test_function()
assert actual == expected
この例では、3つのテストケースを用意しています。ここで用意したパラメータはテスト関数の引数で与えることができ、関数内で用いることができます。これにより、テストを共通化し、わかりやすいテストコードを記述することができます。
pytest-mockを活用する
ユニットテストは単体テストとも呼ばれ、その名の通りコードの中の最小単位(ユニット)をテストするものです。したがって、1つのユニットが複数のユニットに依存している場合、それらのユニットからの影響を無くして、テスト対象のユニットの機能にフォーカスしたテストを作成するべきです。
そのために、ここではpytest-mockを用います。pytest-mockはpytestに完全に統合されているのでpytestが使える環境であれば、すぐに使うことができます。
@pytest.mark.asyncio
async def test_with_mocker(db_session: AsyncSession, mocker):
user_name = "test_user"
user_password = "test_password"
mocker.patch("get_user", return_value=User(id=1, name=user_name))
mocker.patch("get_password", return_value=Password(user_id=1, password=user_password))
mocker.patch("verify_password", return_value=True)
authenticated = await authenticate_user(db_session, user_name, user_password)
assert authenticated
上の例にもあるようにmockerを関数の引数として与えて、mocker.patch(関数名,return_value=value)という形で利用できます。
今回は、authenticate_userをテストしたいので、その中で呼び出される関数についてはmockerを利用して値を返すようにしています。
また、単体テストを作る上で、実行時間をなるべく短くすることが重要です。この観点からも、DBを用いてW/Rを行う処理や、外部APIを呼び出す等の時間がかかる処理も短縮することができます。
単体テストのベストプラクティス
これはFastAPIのテストに限った話ではないのですが、知らなかったテストのベストプラクティスが多くあったので、特に重要だと思った項目をいくつかピックアップしようと思います。
- テストの命名規則(関数名_状況_結果)を守る
- テスト名から何をテストしているのか、どのような条件下で実行されるのか、そして期待される結果は何かを一目で理解できます。これにより、テストコードを読む際にその内容を素早く把握できます
- AAA配置パターン(Arange-Act-Assert)
- テストコードの各部分が明確に分離されているため、他の開発者がコードを読んだときに、何がテストされているのかを簡単に理解できます
他にも勉強になる点が多くあったので、あまりテストコードを書いたことがない方は、ぜひ以下のサイトを参考にしてみてください。
ユニットテスト
conftest.py
ASYNC_DB_URL = "mysql+aiomysql://root:rootpassword@db:3306/data?charset=utf8"
class AsyncTestingSession(AsyncSession):
def commit(self):
self.flush()
self.expire_all()
@pytest.fixture(scope="function")
async def db_session():
async_engine = create_async_engine(ASYNC_DB_URL, echo=True)
async_session = sessionmaker(
autocommit=False, autoflush=False, bind=async_engine, class_=AsyncSession
)
async with async_session() as session:
try:
yield session
finally:
await session.rollback()
await session.close()
DBの前処理について
今回は、テストの前段階でDBの非同期セッションを用意し、テスト終了後にロールバックを行い、セッションをクローズしています。
commitの処理について
ここで、テストを行う際にコミットされたデータを永続化しないために、AsyncSessionを継承し、commitメソッドをオーバーライドすることでコミットを実行する代わりにデータをフラッシュして全てのオブジェクトの状態をリセットしています。
補足
@pytest.fixture(scope="function")のscopeはfixtureのライフサイクルを制御するものです。特に指定しない場合はfunctionが適用されます。これ以外にもsession(pytestの起動から終了まで適用)やclass(テストクラスごとに適用)などがあります。
実際のテスト
テスト対象
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
テスト
@pytest.mark.parametrize(
["test_input", "expected"],
[
pytest.param({"correct_password": "test_password", "input_password": "test_password"}, True),
pytest.param({"correct_password": "test_password", "input_password": "wrong_password"}, False)
]
)
def test_verify_password_verify_input_password_with_hedhed_password_return_verified(test_input, expected):
#Arrange
correct_password = test_input["correct_password"]
input_password = test_input["input_password"]
hashed_password = auth_modules.get_password_hash(correct_password)
#Act
verified = auth_modules.verify_password(input_password, hashed_password)
#Assert
assert verified == expected, "Password verification failed"
APIテスト
conftest.py
TEST_DB_URL = "mysql+aiomysql://root:rootpassword@db:3306/test_db?charset=utf8"
@pytest.fixture(scope="function")
async def async_client() -> AsyncClient:
async_engine = create_async_engine(TEST_DB_URL, echo=True)
async_session = sessionmaker(
autocommit=False, autoflush=False, bind=async_engine, class_=AsyncSession
)
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
async def get_db_for_testing():
async with async_session() as session:
yield session
app.dependency_overrides[get_db] = get_db_for_testing
async with AsyncClient(app=app, base_url="http://localhost:8000") as client:
yield client
使用するDB
APIテストでは実際にテストクライアントを作成してAPIエンドポイントにリクエストを送ることでAPIのテストを実施します。そのため、開発で用いているDBに接続してしまうとそのDBに影響が出てしまう恐れがあるので、別のDBを用意してテストを行っています。公式ドキュメントや他の記事ではSQLiteを用いてAPIテストを行っていることが多いですが、個人的には実際に使用しているDBを用いてテストを行うのが良いと思います。
DBの前処理について
ここではユニットテストと同様にDBセッションの作成を行い、非同期エンジンを使ってデータベース接続を開始し、テスト用のデータベーススキーマをドロップ、再作成します。これにより、各テストが実行される前にクリーンな状態のデータベースが用意されます。
依存関係注入（Dependency Injection, DI）のオーバーライド
FastAPIにはDependsというDIを行うクラスが用意されています。DB接続部分にDIを利用することにより、ビジネスロジックとDBを疎結合に保ってくれます。また、DIによってこの依存関係をoverrideすることが可能なのでテストを行うときのみテスト用のDBに上書きすることができます。
非同期クライアントの作成
非同期処理を行う関数を含むAPIのテストを行うので、最後にテスト用のクライアントを作成してそのクライアントをそれぞれのテストに渡します。
実際のテスト
テスト対象
@router.post("/auth/register")
async def register_new_account(
auth_body : auth_schema.UserCreate, db: AsyncSession = Depends(get_db)
):
new_user = await auth_cruds.create_user(db, auth_body.user_name)
await db.commit()
await db.refresh(new_user)
password_body = auth_schema.PasswordCreate(user_id=new_user.id, password=auth_body.password)
new_password = await auth_cruds.create_password(db, password_body)
await db.commit()
return {"message" : "user created."}
テスト
@pytest.mark.asyncio
async def test_register_new_account_register_user_return_success_message(async_client):
response = await async_client.post("/auth/register", json={"user_name": "test_user", "password": "test_password"})
assert response.status_code == 200
assert response.json() == {"message": "user created."}
継続的インテグレーション（CI）の実装
pytestはもちろんローカルでも実行することはできますが、CIとして開発のフローの中に入れてしまえば、テストを意識せずとも自動でテストを実行してくれますし、リファクタリング等を行った際に早い段階でミスに気づき、修正を行うことができます。テストを作成したらぜひCIまで作成してみてください。
ユニットテストのymlファイル
ここでは、mainブランチにpull requestが作成された時に自動的にテストを行うようにしています。まずリポジトリをチェックアウトしdockerコンテナを立ち上げています。その後、コンテナの中に入り、dependencyのインストールを行い、DBのマイグレーションを行うことでDBのセットアップを行っています。最後にテストを実行し、その結果をPRのコメント欄に記述しています。
PRへテスト結果のコメントを残すために最低限の権限を与えています。コメントを残さなくてもいいという方はこの部分はなくても大丈夫です。
name: Run Unit Tests on Pull Request
on:
pull_request:
branches:
- main
permissions:
contents: read
pull-requests: write
jobs:
pytest_coverage:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Build and Start Services
run: |
docker-compose build
docker-compose up -d
sleep 20
- name: Run Tests
run: |
docker container exec back poetry install --no-root
docker container exec back poetry run python -m migrate_db
docker container exec back poetry run pytest test/unit_test --asyncio-mode=auto --cov --cov-branch --cov-report=term-missing --junitxml=pytest.xml test | tee pytest-coverage.txt
- name: Pytest coverage comment
uses: MishaKav/pytest-coverage-comment@main
with:
pytest-coverage-path: ./pytest-coverage.txt
junitxml-path: ./pytest.xml
APIテストのymlファイル
ユニットテストと同様にmainブランチにpull requestが作成された時に自動的にテストを行うようにしています。まずリポジトリをチェックアウトしdockerコンテナを立ち上げています。その後、コンテナの中に入り、dependencyのインストールを行い、DBのマイグレーションを行うことでDBのセットアップを行っています。最後にテストを実行し、その結果をPRのコメント欄に記述しています。
name: Run API Tests on Pull Request
on:
pull_request:
branches:
- main
permissions:
contents: read
pull-requests: write
jobs:
pytest_coverage:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Build and Start Services
run: |
docker-compose build
docker-compose up -d
sleep 20
- name: Run Tests
run: |
docker container exec back poetry install --no-root
docker container exec back poetry run python -m migrate_test_db
docker container exec back poetry run pytest test/integration_test --asyncio-mode=auto --cov --cov-branch --cov-report=term-missing --junitxml=pytest.xml test | tee pytest-coverage.txt
- name: Pytest coverage comment
uses: MishaKav/pytest-coverage-comment@main
with:
pytest-coverage-path: ./pytest-coverage.txt
junitxml-path: ./pytest.xml
補足
pytestは引数にテストのパスを与えることで特定のテストのみを実行することができます。したがって、それぞれのworkflowで実行したいテストのみを実行しています。
補足
MySQLを使ってテストを行う方は、コンテナを立ち上げる際に/docker-entrypoint-initdb.d/init.sqlにあるSQLが実行されるので、そこにテスト用DBを作成するSQLを記述すればコンテナ起動時にDBが追加で作られます。
参照：https://hub.docker.com/_/mysql
テスト結果
テストが終了するとPRのコメント欄に自動的にテスト結果とカバレッジ率を表示してくれます。これで、コードのどの部分のテストが行われていないのかすぐにわかります。
補足
これは、pytestの実行結果をpytest-coverage.txtとpytest.xmlに書き出し、GitHub Marketplaceで提供されているアクションMishaKav/pytest-coverage-comment@mainを使用してコメントにテスト結果を書き出しています。
おわりに
テストを書くのは面倒ですが、テストを書いて、CIを実装することでコードの品質を維持し、安全な変更を迅速にデプロイすることができます。少しでもテスト作成の参考になれば嬉しいです。ここまで読んでいただきありがとうございました。
ご指摘やアドバイスがあればコメントで教えていただけると幸いです。