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 ==========================================