・5 min read
【開発】pytest×SQLAlchemyでテストする方法
なにこれ
SQLAlchemyを利用しているプログラムをpytestを使ってテストするための実装方法をまとめたものです。
システム構成
テスト対象となるアプリケーションには、開発・STG・本番環境などの運用DBの他にテスト実行時に利用するテストDBを環境を用意しておきます。テストDBはDockerコンテナ上に構築するなりしてください。
アプリケーション稼働時のコネクション先とpytest実行時のコネクション先はそれぞれ以下のようになります。
アプリケーション稼働時のコネクション
pytest実行時のコネクション
ディレクトリ構成
- アプリケーションコードとテストコードのディレクトリは同じ構成になるように定義してあります。
- conftest.pyにテストDBに接続するフィクスチャを定義します。
.
├── app # アプリケーションコード
│ ├── __init__.py
│ └── package1
│ ├── __init__.py # 運用DBとの接続/セッションを宣言するファイル
│ └── users_table.py # テーブル定義しているモジュール
└── tests # テストコード
├── __init__.py
├── conftest.py # 全テストで参照可能なフィクスチャを定義するファイル
└── package1
├── __init__.py
└── test_users_table.py # sample_table.pyに対応するテストコードコード一覧
app
app/package1/__init__.py
import os
from sqlalchemy import create_engine
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import declarative_base, sessionmaker
# Baseを継承して、テーブルを定義します
Base = declarative_base()
# 運用DBに接続し、セッションを確立します
engine = create_engine(
URL.create(
drivername=os.environ.get('DB_DRIVERNAME'),
username=os.environ.get('DB_USERNAME'),
...,
database='db' # 運用DB
)
)
session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))app/package1/users_table.py
from sqlalchemy import Column, VARCHAR
from app.package1 import Base
class UsersTable(Base):
__tablename__ = 'users'
id = Column(VARCHAR(255), primary_key=True, nullable=False)
name = Column(VARCHAR(255), nullable=False)tests
tests/conftest.py
import pytest
from sqlalchemy.orm import Session, scoped_session, sessionmaker
from package1 import Base
@pytest.fixture(scope="session")
def connection():
engine = create_engine(
URL.create(
drivername=os.environ.get('DB_DRIVERNAME'),
username=os.environ.get('DB_USERNAME'),
...,
database='testdb' # テストDBを指定
)
)
# テストDBにコネクションを確立する
return engine.connect()
@pytest.fixture(scope="session")
def setup_database(connection):
# 前処理：全テーブルを作成
Base.metadata.bind = connection
Base.metadata.create_all()
yield
# 後処理：テーブルを全て削除
Base.metadata.drop_all()
@pytest.fixture
def test_session(setup_database, connection) -> Session:
transaction = connection.begin()
# テスト用のセッションを作成
yield scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=connection))
transaction.rollback()tests/package1/test_users_tablepy
from sqlalchemy.orm import Session
from app.package1.users_table import UsersTable
class Test_UsersTable:
def test_テストコード(self, test_session: Session):
# テストデータを保存
test_session.add(UsersTable(id='a', name='a'))
test_session.commit()
assert test_session.query(UsersTable).get('a')