LoginSignup
3
0

More than 1 year has passed since last update.

【FastAPI】テストケース毎に独立したDBデータを使用する(GitHub Actions付き)

Posted at

ORMと連携した機能のテストを行う際、テストケース(= テスト関数)毎にクリーンなデータベースが欲しい(テストケース間の依存関係が生まれてほしくない)。

この点についていい感じの方法が実現できたためご紹介する。

本記事のソースコード:skokado/fastapi-tutorial

環境

  • Python: 3.8
    • fastapi==0.68.2
    • SQLAlchemy==1.4.25
    • SQLAlchemy-Utils==0.37.8
    • pytest==6.2.5
    • factory-boy==3.2.0

アプリケーション準備

ユーザ認証とブログ管理を行う簡単なアプリケーションを用意する。
※アプリケーション本体のコードは割愛するためリポジトリを参照

※ディレクトリ構成
.
├── Pipfile
├── Pipfile.lock
├── README.md
├── alembic.ini
├── app
│   ├── __init__.py
│   ├── core
│   │   ├── __init__.py
│   │   ├── celery_app.py
│   │   ├── config.py
│   │   └── security.py
│   ├── crud
│   ├── database.py
│   ├── dependencies.py
│   ├── main.py
│   ├── middlewares.py
│   ├── models
│   ├── routers
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── blogs.py
│   │   └── users.py
│   ├── schemas.py
│   ├── tests
│   │   ├── ...
│   └── utils
│       ├── oauth2.py
│       └── token.py
└── migration
│   ├── README
│   ├── env.py
│   ├── script.py.mako
│   └── versions
│       ├── ...

本題

データベース準備

DockerコンテナでPostgreSQLを起動しておく

$ docker run -d --rm \
   --name test_db \
  -p 5433:5432 \
  -e POSTGRES_USER=app \
  -e POSTGRES_PASSWORD=secret \
  -e POSTGRES_DB=app \
  postgres:13.4

フィクスチャ準備

テスト用データベースを使用する設定をconftest.pyに仕込んでおき、データベースをfixtureとして使用できるようにしておく。

※こちらのGitHub Issueも参照されたし
Separating database for tests and dev · Issue #125 · tiangolo/full-stack-fastapi-postgresql

app/tests/conftest.py
app/tests/conftest.py
from typing import Generator
import pytest
from sqlalchemy.orm import Session
from sqlalchemy_utils import database_exists, create_database

from app.core.config import settings
from app.database import Base
from tests.utils.overrides import TestingSessionLocal, engine


@pytest.fixture(scope="function")
def db() -> Generator[Session, None, None]:
    if not database_exists(settings.SQLALCHEMY_DATABASE_URI):
        create_database(settings.SQLALCHEMY_DATABASE_URI)

    Base.metadata.create_all(bind=engine)
    session = TestingSessionLocal()
    yield session
    session.close()
    Base.metadata.drop_all(bind=engine)

ポイントは、dbフィクスチャのスコープを"function"としている点。
こうすればテスト関数毎にセッションをクローズ&データベース初期化が行われ、データがクリーンアップされる。

app/tests/conftest.py
...
    session = TestingSessionLocal()
    yield session
    session.close()
    Base.metadata.drop_all(bind=engine)

テスト用データ作成

factory-boyを使いFactoryクラスを作成しておく。
テストケースの中で簡単にインスタンス化できるようにする。

app/tests/factories.py
app/tests/factories.py
from factory.alchemy import SQLAlchemyModelFactory

import app.models.user as models
from app.tests.utils.overrides import TestingSessionLocal


class UserFactory(SQLAlchemyModelFactory):
    name = 'skokado'
    email = 'skokado@example.com'
    password = 'MyPassw0rd!'

    class Meta:
        model = models.User
        sqlalchemy_session = TestingSessionLocal()

テストケース

テストケースの中でdbフィクスチャを使用してデータ作成をするのみで良い。

app/tests/routers/test_auth.py
app/tests/routers/test_auth.py
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app.main import app
import app.crud.users as users_crud
from app.tests import factories

client = TestClient(app)


class TestAuthRouter:

    def test_ログイン成功(self, db: Session):
        user_in = factories.UserFactory()
        users_crud.create(user_in, db)
        response = client.post(
            '/api/login',
            data={
                'grant_type': 'password',
                'username': user_in.name,
                'password': user_in.password
            }
        )
        assert response.status_code == 200

app/tests/routers/test_blogs.py
app/tests/routers/test_blogs.py
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app.main import app
import app.crud.users as users_crud
from app.core.security import create_access_token
from app.tests import factories

client = TestClient(app)
client.headers.update({
    'Authorization': 'Bearer {}'.format(create_access_token)
})


class TestBlogRouter:
    def test_list_empty_blogs(self, db: Session):
        user_in = factories.UserFactory()
        users_crud.create(user_in, db)
        # ログインする
        response = client.post(
            '/api/login',
            data={
                'grant_type': 'password',
                'username': user_in.name,
                'password': user_in.password
            }
        )
        jwt = response.json()
        headers = {
            'Authorization': f'{jwt["token_type"].capitalize()} {jwt["access_token"]}'
        }

        # アクセストークンを使用してAPIリクエスト
        response = client.get('/api/blog/', headers=headers)
        assert response.status_code == 200
        assert response.json() == []

3
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
3
0