Magicode logo
Magicode
0
5 min read

【開発】pytest×SQLAlchemyでテストする方法

https://cdn.apollon.ai/media/notebox/437b0249-b67a-4472-b039-b26ff4e9b75b.jpeg

なにこれ

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')

Discussion

コメントにはログインが必要です。