sqlalchemyを使った実装をしていて、単体テストをする際、
単純なクエリであればmockを使用したテストでも良いかと思いますが、
複雑なクエリをクエリビルダーによって構成している場合は、
それらをmockで確認するのはちょっと現実的ではないと思います。
そこで、今回はローカル環境にdockerを立ててそこにDBを構築し、
実際にSQLを実行した結果を元に単体テストする方法についてご紹介したいと思います。
データベースはPostgresqlです。
DB接続関連の実装
前提として、以下のようにDB接続部分を実装しているものとします。
接続情報(settings.py)
接続情報は環境変数に設定されているものとして、以下のように取得します。
|
1 2 3 4 5 6 7 8 9 |
"""定数モジュール""" 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を使用した接続のセッション管理周りはこちらで実装しているものとします。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
"""データベースセッション管理モジュール""" 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として環境準備を行います。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 |
# 単体実行時は以下をコメントアウトする @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で扱うためには、
|
1 |
pip install docker |
でインストールしておきます。
コンテナ起動後は、すぐにアクセスできないので
ちょっとだけ暖気運転させます。
その後はSQLファイルを読み込んで実行させ、
環境を整えます。
あとは、db_initのfixtureを引数としたテストを実装するという流れになります。
まとめ
今回の方法はjenkinsなどのCIツールで回そうとすれば、
dockerの起動を必須としますので、環境によってはサンプルコードのように、
skipのマークをつけておくのがよいかもしれないですねー。
20代前半までは東京で音楽をやりながら両手の指以上の業種でアルバイト生活をしていましたが、某大手プロバイダのテレアポのバイトでPCの知識の無さに愕然とし、コンピュータをもっと知りたい！と思ったことをきっかけに25歳の時にITの世界に未経験で飛び込みました。
紆余曲折を経て、現在は個人事業主としてお仕事させていただいており、10年ほどになります。
web制作から企業システム構築、ツール開発など、フロントエンドもバックエンドもサーバーもDBAも依頼があれば何でもやってきた雑食系エンジニアです。
今風にいうとフルスタックエンジニアということになるのかしら？？
→ 詳細プロフィールというか、生い立ちはこちら
→スキルシートをご覧になる場合はこちら
→お仕事のご依頼やお見積りなどお問い合わせはこちらから！