【python】【sqlalchemy】ローカル環境でdockerにデータベースを立てて単体テストする

sqlalchemyを使った実装をしていて、単体テストをする際、

単純なクエリであればmockを使用したテストでも良いかと思いますが、

複雑なクエリをクエリビルダーによって構成している場合は、

それらをmockで確認するのはちょっと現実的ではないと思います。

 

 

そこで、今回はローカル環境にdockerを立ててそこにDBを構築し、

実際にSQLを実行した結果を元に単体テストする方法についてご紹介したいと思います。

データベースはPostgresqlです。

 

 

 

 

 

 

 

 

DB接続関連の実装

前提として、以下のようにDB接続部分を実装しているものとします。

接続情報(settings.py)

接続情報は環境変数に設定されているものとして、以下のように取得します。

"""定数モジュール"""
import os

# データベース接続情報
DB_USER = os.environ["DBUSER"]
DB_PASS = os.environ["DBPASS"]
DB_HOST = os.environ["DBHOST"]
DB_PORT = os.environ["DBPORT"]
DB_NAME = os.environ["DBNAME"]

 

 

接続処理(db.database.py)

sqlalchemyを使用した接続のセッション管理周りはこちらで実装しているものとします。

"""データベースセッション管理モジュール"""
from contextlib import contextmanager

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from settings import DB_HOST, DB_NAME, DB_PASS, DB_PORT, DB_USER

# 接続URI
DATABASE_URI = (
    f"postgresql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
)

# メタクラス
DECLARATIVE_BASE = declarative_base()

# engine生成
engine = create_engine(
    DATABASE_URI,
    encoding="utf-8",
    echo=True,
)

# session生成
base_session = sessionmaker(autocommit=False, autoflush=False, bind=engine)


@contextmanager
def DBController():
    """DATABASEコネクション・セッション操作"""
    session = base_session()
    yield session
    if session:
        session.close()

 

withを用いて使用するものは、contextmanagerを使用した書き方が

良いかと思います。詳しくはこちら。

 

 

単体テスト実装

単体テストではfixtureとして環境準備を行います。

# 単体実行時は以下をコメントアウトする
@pytest.mark.skip(reason="docker boot required")
class TestDbQuery:
    """クエリテスト
    dockerでPostgresqlを立ててクエリ実行する
    """

    @pytest.fixture
    def mock_before_import(self, mocker):
        mocker.patch.dict(
            os.environ,
            {
                "DBUSER": "postgres",
                "DBPASS": "testpass",
                "DBHOST": "localhost",
                "DBPORT": "45432",
                "DBNAME": "postgres",
            },
        )

    @pytest.fixture(scope="class")
    def postgres(self):
        client = docker.from_env()
        container = client.containers.run(
            "postgres:9.6",
            detach=True,
            environment=["POSTGRES_PASSWORD=testpass"],
            ports={"5432/tcp": 45432},
        )

        # docker内のpostgresqlが再起動するので安定するのを待つ
        for _ in range(100):
            try:
                from sqlalchemy import create_engine

                engine = create_engine(
                    "postgresql://postgres:testpass@localhost:45432"
                )
                engine.execute("CREATE SCHEMA IF NOT EXISTS test;")
            except Exception:
                time.sleep(0.1)
            else:
                break
        yield
        container.stop(timeout=0)

    @pytest.fixture
    def db_init(self, mock_before_import, postgres):
        """dokerのpostgres_dbにテーブル、マスタ、テストデータを設定."""
        # テーブルを作成
        from db.database import DBController

        with DBController() as session:

            session.execute("CREATE SCHEMA IF NOT EXISTS test")

            top_dir = os.path.dirname(__file__)
            file_path = os.path.join(top_dir, "./scripts/create_table.sql")

            with open(file_path, encoding="UTF-8") as f:
                session.execute(f.read())

            file_path = os.path.join(top_dir, "./scripts/insert_data.sql")

            with open(file_path, encoding="UTF-8") as f:
                session.execute(f.read())

            session.commit()

 

環境変数についてはpytest-mockを使用して、設定します。

dockerについては事前に起動しておく必要があり、pythonで扱うためには、

pip install docker

でインストールしておきます。

 

コンテナ起動後は、すぐにアクセスできないので

ちょっとだけ暖気運転させます。

 

その後はSQLファイルを読み込んで実行させ、

環境を整えます。

 

 

あとは、db_initのfixtureを引数としたテストを実装するという流れになります。

 

まとめ

今回の方法はjenkinsなどのCIツールで回そうとすれば、

dockerの起動を必須としますので、環境によってはサンプルコードのように、

skipのマークをつけておくのがよいかもしれないですねー。

 

 

にほんブログ村

人気ブログランキング