Pytestのテスト実行前にDockerコンテナを起動することを考える。
周辺ミドルウェアとのインテグレーションテストが容易に行えるようになり、環境やテストデータや冪等性を得られるメリットがある。
例としてテスト実行時にPostgreSQLコンテナを起動し、データを投入してみる。
備考
コード例の前に、ユニットテストの中でDockerコンテナを使用することの制約などについて考えてみる。
-
Dockerデーモンが起動している必要がある
-> 当然といえば当然だが重要。コンテナの中でテスト実行したいようなケースにはハードルが1つ増える(?)- Docker in Dockerができるコンテナをビルドすればできそう（未検証）。
- 参考: Docker in Docker のベタープラクティス
-
コンテナ起動まで待機する必要あり
- コンテナ自体の起動は高速だが、内部のプロセスが起動完了して使用可能になるまでは数秒かかる場合が多い。
- テスト関数が増えてくると全テストケース実行時間が大幅に増加してしまう
- なるべく軽量なコンテナイメージを使う方が良い
- コンテナの標準出力をウォッチすることで準備完了になるまで適切に待機するよう工夫する
使用ライブラリ
- pytest: ユニットテスト
- docker: Docker APIのラッパー - docker - PyPI
- SQLAlchemy: ORM
- psycopg2: PostgreSQL接続ドライバ
$ pip install docker pytest SQLAlchemy psycopg2-binary
構成
├── main.py
├── models.py
└── tests
├── __init__.py
├── conftest.py
└── test_main.py
メイン処理/DBモデル
- models.py
models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Staff(Base):
__tablename__ = 'staff'
id = Column(Integer, primary_key=True)
name = Column(String)
- main.py
main.py
from sqlalchemy.orm import Session
from models import Staff
def add_staff(engine, name):
session = Session(bind=engine)
staff = Staff()
staff.name = name
session.add(staff)
session.commit()
テストコード
SQLAlchemyのengineをフィクスチャにすることでテスト関数で使用できるようにするのがポイント。
- conftest.py
tests/conftest.py
import time
import pytest
import docker
from sqlalchemy import create_engine
@pytest.fixture()
def pg_conf():
"""PostgreSQLの設定を管理"""
host = '127.0.0.1'
port = 5432
dbname = 'pytest'
user = 'testuser'
password = 'test'
pg_conf = {'host': host,
'port': port,
'dbname': dbname,
'user': user,
'password': password,
'url': f'postgresql://{user}:{password}@{host}/{dbname}'}
return pg_conf
@pytest.fixture()
def engine(pg_conf):
return create_engine(pg_conf['url'])
@pytest.fixture(autouse=True)
def pg_container(pg_conf):
"""PostgreSQLコンテナを起動する"""
client = docker.from_env()
container = client.containers.run(image='postgres:11.6-alpine',
tty=True,
detach=True,
auto_remove=True,
environment={'POSTGRES_DB': pg_conf['dbname'],
'POSTGRES_USER': pg_conf['user'],
'POSTGRES_PASSWORD': pg_conf['password']},
ports={pg_conf['port']: '5432'})
# コンテナが準備完了になるまで待機
while True:
log = container.logs(tail=1)
if 'database system is ready to accept connections' in log.decode():
break
time.sleep(0.5)
yield # ここでテストに遷移
container.kill()
コンテナ内の標準出力をチェックしているが、待機間隔が短すぎる（0.4秒間隔以下）とエラーが発生してしまった。
少し猶予を持った待機時間にしたほうが良さそう。
- test_main.py
test_main.py
from sqlalchemy.orm import Session
from models import Base, Staff
from main import add_staff
def test_add(engine):
# レコードを1件追加
Base.metadata.create_all(bind=engine) # テーブルを作成
add_staff(engine=engine,
name='alice')
# 追加したレコードをチェック
session = Session(bind=engine)
assert session.query(Staff.id).filter_by(name='alice').first() == (1,)
session.close()
実行結果
$ pytest --setup-show tests/ -v -s
========================================= test session starts =========================================platform linux -- Python 3.8.1, pytest-5.3.3, py-1.8.1, pluggy-0.13.1 -- /home/skokado/.local/share/virtualenvs/sandbox-pTebjwBw/bin/python3.8
cachedir: .pytest_cache
rootdir: ***
collected 1 item
tests/test_pg.py::test_add
SETUP F pg_conf
SETUP F pg_container (fixtures used: pg_conf)
SETUP F engine (fixtures used: pg_conf)
tests/test_main.py::test_add (fixtures used: engine, pg_conf, pg_container)PASSED
TEARDOWN F engine
TEARDOWN F pg_container
TEARDOWN F pg_conf
========================================== 1 passed in 2.00s ==========================================