Python で Testcontainers を使い始める

コードを入手する

Testcontainers for Python は、開発やテストの際に使い捨て可能なオンデマンドのコンテナを提供するオープンソースのフレームワークです。これにより、データベース、メッセージブローカー、ウェブブラウザ、その他 Docker コンテナで実行できるほぼすべてのものと簡単に連携できます。Testcontainers を使用すると、モックやインメモリサービスを使用せずに、本番環境と同じ種類のサービスと対話するテストを作成できます。

Testcontainers を初めて使用する場合は、Testcontainers とは何か、なぜ使うべきなのか？を読んで、Testcontainersについて詳しく学んでください。

以下の手順に従って、PostgreSQL データベースを使用するシンプルな Python アプリケーションを作成し、Testcontainers を使用して実際の PostgreSQL データベースでテストを行う方法を学びましょう。

Python アプリケーションの作成

まず、Python プロジェクトを作成し、venv モジュールを使用して仮想環境を設定します。仮想環境を使用することで、依存関係をグローバルにインストールすることを避け、プロジェクトごとに異なるバージョンのパッケージを使用できます。

$ mkdir tc-python-demo
$ cd tc-python-demo
$ python3 -m venv venv
$ source venv/bin/activate

次に、PostgreSQL データベースと対話するために psycopg3、テストのために pytest、コンテナ内で PostgreSQL データベースを実行するために testcontainers-python を使用します。仮想環境をアクティブにした状態で、以下のコマンドで必要な依存関係をインストールします。

$ pip install psycopg pytest testcontainers[postgres]
$ pip freeze > requirements.txt

依存関係をインストールしたら、pip freeze コマンドを使用して requirements.txt ファイルを生成します。これにより、他の人が同じバージョンのパッケージを簡単にインストールできます。

データベースヘルパーの実装

db/connection.py ファイルを作成し、データベース接続を取得する関数を以下のように定義します。

import os
import psycopg
 
def get_connection():
    host = os.getenv("DB_HOST", "localhost")
    port = os.getenv("DB_PORT", "5432")
    username = os.getenv("DB_USERNAME", "postgres")
    password = os.getenv("DB_PASSWORD", "postgres")
    database = os.getenv("DB_NAME", "postgres")
    return psycopg.connect(f"host={host} dbname={database} user={username} password={password} port={port}")

データベース接続パラメータをハードコーディングする代わりに、環境変数を使用して取得しています。これにより、コードを変更せずに異なる環境でアプリケーションを実行できます。

ビジネスロジックの実装

customers/customers.py ファイルを作成し、Customer クラスを以下のように定義します。

class Customer:
    def __init__(self, cust_id, name, email):
        self.id = cust_id
        self.name = name
        self.email = email
 
    def __str__(self):
        return f"Customer({self.id}, {self.name}, {self.email})"

次に、customers/customers.py ファイルに create_table() 関数を追加し、customers テーブルを作成します。

from db.connection import get_connection
 
def create_table():
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE customers (
                    id serial PRIMARY KEY,
                    name varchar not null,
                    email varchar not null unique)
                """)
            conn.commit()

get_connection() 関数を使用して新しいデータベース接続を取得し、customers テーブルを作成しています。with ステートメントを使用して、データベース接続を自動的に閉じるようにしています。

さらに、customers/customers.py ファイルに以下の関数を実装します。

create_customer(name, email): 新しい顧客を作成します。
get_all_customers(): すべての顧客を取得します。
get_customer_by_email(email): メールアドレスで顧客を検索します。
delete_all_customers(): すべての顧客を削除します。

def create_customer(name, email):
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO customers (name, email) VALUES (%s, %s) RETURNING id
                """, (name, email))
            customer_id = cur.fetchone()[0]
            conn.commit()
            return Customer(customer_id, name, email)
 
def get_all_customers():
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id, name, email FROM customers")
            rows = cur.fetchall()
            return [Customer(row[0], row[1], row[2]) for row in rows]
 
def get_customer_by_email(email):
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id, name, email FROM customers WHERE email = %s", (email,))
            row = cur.fetchone()
            if row:
                return Customer(row[0], row[1], row[2])
            return None
 
def delete_all_customers():
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM customers")
            conn.commit()

Python の DB-API を使用して、データベースから顧客のレコードを挿入、取得、削除するためのさまざまな機能を実装しました。

このガイドの目的を簡単にするため、各データベース操作ごとに新しい接続を作成しています。実際のアプリケーションでは、接続プールを使用して接続を再利用することをお勧めします。

Testcontainers を使用したテストの作成

Testcontainers を使用して PostgreSQL データベースコンテナのインスタンスを作成し、すべてのテストで同じデータベースを使用します。また、各テストの前にすべての顧客レコードを削除することで、テストをクリーンなデータベース環境で実行できるようにします。

セットアップとテアダウン（後処理）のロジックを実装するために、pytest の fixtures を使用します。セットアップとテアダウンのロジックを実装する推奨方法は、yield fixtures を使用することです。

@pytest.fixture
def setup():
    # setup code
    yield some_value
    # teardown code

しかし、この方法では、セットアップコード内で例外が発生した場合、テアダウンコードが実行されない可能性があります。そのため、より良いアプローチとして以下のように finalizers を使用する方法があります：

@pytest.fixture
def setup(request):
    # setup code
 
    def cleanup():
        # teardown code
 
    request.addfinalizer(cleanup)
    return some_value

tests/test_customers.py ファイルを作成し、以下のように fixtures を実装しましょう。

import os
import pytest
from testcontainers.postgres import PostgresContainer
 
from customers import customers
 
postgres = PostgresContainer("postgres:16-alpine")
 
 
@pytest.fixture(scope="module", autouse=True)
def setup(request):
    postgres.start()
 
    def remove_container():
        postgres.stop()
 
    request.addfinalizer(remove_container)
    os.environ["DB_CONN"] = postgres.get_connection_url()
    os.environ["DB_HOST"] = postgres.get_container_host_ip()
    os.environ["DB_PORT"] = postgres.get_exposed_port(5432)
    os.environ["DB_USERNAME"] = postgres.username
    os.environ["DB_PASSWORD"] = postgres.password
    os.environ["DB_NAME"] = postgres.dbname
    customers.create_table()
 
 
@pytest.fixture(scope="function", autouse=True)
def setup_data():
    customers.delete_all_customers()

module スコープの fixture を使用して、Testcontainers を用いて PostgreSQL コンテナを起動しています。この fixture は module 内のすべてのテストに対して一度だけ実行されます。setup() fixture 関数では、PostgreSQL コンテナを起動し、customers テーブルを作成しています。また、すべてのテストが終了した後にコンテナを削除するための finalizer を追加しています。

setup_data() fixture 関数では、customers テーブル内のすべてのレコードを削除しています。この関数は関数スコープの fixture であり、各テストの実行前に実行されます。

現在のところ、Testcontainers for Python では Ryuk を使用した自動リソースクリーンアップが実装されていないため、finalizer を使用して明示的に作成したコンテナを削除しています。

それでは、以下のようにテストを実装しましょう：

def test_get_all_customers():
    customers.create_customer("Siva", "siva@gmail.com")
    customers.create_customer("James", "james@gmail.com")
    customers_list = customers.get_all_customers()
    assert len(customers_list) == 2
 
 
def test_get_customer_by_email():
    customers.create_customer("John", "john@gmail.com")
    customer = customers.get_customer_by_email("john@gmail.com")
    assert customer.name == "John"
    assert customer.email == "john@gmail.com"

test_get_all_customers() テストでは、2つの顧客レコードをデータベースに挿入し、既存のすべての顧客を取得して顧客数をアサートします。
test_get_customer_by_email() テストでは、顧客レコードをデータベースに挿入し、メールアドレスで顧客を取得して、その詳細をアサートします。

各テストの実行前にすべての顧客レコードを削除しているため、テストは任意の順序で実行することが可能です。

テストの実行

Pytest の自動検出メカニズムを有効にするために、tests ディレクトリ内に空の内容の init.py ファイルを作成します。

次に、以下のように pytest を使用してテストを実行します：

$ pytest

以下のような出力が表示されるはずです：

pytest
=============== test session starts ==============
platform darwin -- Python 3.12.0, pytest-7.4.3, pluggy-1.3.0
rootdir: /Users/siva/dev/tc-python-demo
collected 2 items
 
tests/test_customers.py ..                       [100%]
 
============== 2 passed in 3.02s =================

これらのテストはモックではなく、実際の PostgreSQL データベースを使用して実行されるため、実装に対する信頼性が向上します。

まとめ

このガイドでは、testcontainers-python ライブラリを使用して PostgreSQL データベースを利用した Python アプリケーションのテスト方法を紹介しました。testcontainers-python は、PostgreSQL のほかにも、一般的に使用される多くの SQL データベース、NoSQL データベース、メッセージキューなどに対応した専用モジュールを提供しています。また、Testcontainers を使用することで、テストのために任意のコンテナ化された依存関係を実行することが可能です！

Testcontainers の詳細については、公式サイト https://testcontainers.com をぜひご覧ください。

Node.js で Testcontainers を使い始める Testcontainersとは何か、なぜ使うべきなのか？