pytest-dockerでデータベースをfixtureする

「データベースをDockerでfixtureするpytest環境例」として次を用意した。

• Qiita記事
• GitHub

こちらの記事はMySQLだったがPostgreSQLでも確認してみた。

なお、データベース接続インターフェースのライブラリーとして、

• MySQLの場合: PyMySQL
• PostgreSQLの場合: psycopg2

を利用したが、他のものを利用する場合もほぼ同じだと思う。

conftest.py の fixture例

pytest-dockerのサンプルでhttpbinの起動待ちしているところ、

def is_responsive(url):
    try:
        response = requests.get(url)
        if response.status_code == 200:
            return True
    except ConnectionError:
        return False


@pytest.fixture(scope="session")
def http_service(docker_ip, docker_services):
    """Ensure that HTTP service is up and responsive."""

    # `port_for` takes a container port and returns the corresponding host port
    port = docker_services.port_for("httpbin", 80)
    url = "http://{}:{}".format(docker_ip, port)
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_responsive(url)
    )
    return url

これのデータベース版。アプリケーション側ではSQLAlchemy等のORMを使う場合でも直接接続ライブラリーでconnectしてみた。ORM経由だとSQLAlchemyのcreate_engine()のタイミングなのかどうなのかよくわからないし。ドライバーライブラリー経由で接続して起動確認してるってことを明示する感じで。

MySQL container の場合

MySQLコンテナーが立ち上がりconnect()に対してExceptionがでなくなるまで待つ

def is_mysqld_ready(docker_ip):
    try:
        pymysql.connect(
            host=docker_ip,
            user=os.getenv('MYSQL_USER', ''),
            password=os.getenv('MYSQL_PASSWORD', ''),
            db=os.getenv('MYSQL_DATABASE', '')
        )
        return True
    except:
        return False

@pytest.fixture(scope="session")
def database_service(docker_ip, docker_services):
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_mysqld_ready(docker_ip)
    )
    return

PostgreSQL container の場合

一緒ですね。

def is_postgresql_ready(docker_ip):
    try:
        psycopg2.connect(
            "postgresql://{user}:{password}@{host}/{db}".format(
                user=os.getenv('POSTGRES_USER', ''),
                password=os.getenv('POSTGRES_PASSWORD', ''),
                host=docker_ip,
                db=os.getenv('POSTGRES_DB', '')
            )
        )
        return True
    except:
        return False

@pytest.fixture(scope="session")
def database_service(docker_ip, docker_services):
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_postgresql_ready(docker_ip)
    )
    return

docker-compose.yml

# MySQLの場合
version: "3"

services:
  database:
    image: mysql:5.7
    ports:
      - 3306:3306
    volumes:
      - ./initdb.d:/docker-entrypoint-initdb.d
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
      - MYSQL_USER=${MYSQL_USER}
      - MYSQL_PASSWORD=${MYSQL_PASSWORD}
      - MYSQL_DATABASE=${MYSQL_DATABASE}

# PostgreSQLの場合
version: "3"

services:
  database:
    image: postgres:13
    ports:
      - 5432:5432
    volumes:
      - ./initdb.d:/docker-entrypoint-initdb.d
    environment:
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=${POSTGRES_DB}

スキーマとテストデータの準備

あとはdocker-composeでヴォリュームマウントされ初期化実行される./initdb.d:/docker-entrypoint-initdb.dの./initdb.dフォルダにSQLや初期化シェルスクリプトなどを放り込んでおけばテスト可能に。

Qiitaの記事だと、Docker で MySQL 起動時にデータの初期化を行う など。