Redshiftを使用するプログラムの自動テストの工夫

AWSのAmazon Redshiftを読み書きするPythonプログラムのテストを自動化してみた
そのときの工夫の記録

使用した環境

• Amazon Redshift
• Python3.7
• pytest>=5.3.2
• pandas>=1.0.0
• psycopg2>=2.8.4

データベースとしてRedshiftを使用しているが、他のデータベースプロダクトでも同じ方式で工夫できると思われる

実現したかったこと

プログラムがデータベース(Redshift)に書き込んだ結果データを検証したかった
つまりDAOを使用せずデータベースに接続した状態で、Java/JunitのDBUnitのようなことをしたかった

自動テスト方式

自動テストの流れ

以下のフェーズがある

1. データ退避
2. テストデータ投入
3. テスト実行
4. 結果検証
5. データ復元

これを実現するために今回は主にpytestのfixture関数を作成した

フェーズ詳解

1. データ退避/5. データ復元

Redshiftを読み書きするテスト対象プログラム(Python)をpytest実行する前に、Redshiftに保存されているデータが壊れさないよう全データをS3に退避する
pytest実行が一通り完了したら全データを復元し、Redshiftをpytest実行前の状態にする

実装例

import pytest

# 'session'を指定することで一連のpytest実行前後で処理できるようにする
@pytest.fixture(scope='session')
def fixture_session():
    # Redshiftから全データをS3にUNLOADする処理

    yield

    # 全データをS3からRedshiftにCOPYする処理

pytestを途中でキャンセルしてもyield以降は処理されるため、安全にデータは復元される
すでにRedshiftに大量のデータが存在する場合、退避と復元に時間がかかりpytest実行がスムーズでなくなる

Spectrumのケース

• S3にある実体のデータオブジェクトをそのままのプリフィクス/パーティション構造で別のS3パスへ移動し退避する
• SpectrumテーブルのS3パスは"SVV_EXTERNAL_TABLES"テーブルの"location"カラムから取得する
• S3オブジェクトを移動する際に、S3オブジェクトのACL権限がデフォルト設定になってしまうので注意

2. テストデータ投入/3. テスト実行

テスト関数(テスト対象プログラム)を実行するたび実行前にRedshiftの全データを削除し、Redshiftに残ったテストデータをクリーンする
テスト関数の実行前に、使用するテストデータをRedshiftを投入する

実装例

import pytest

# 'function'を指定することでテスト関数実行前に処理できるようにする
@pytest.fixture(scope='function')
def fixture_function():
    # Redshiftから全データを削除する処理

@pytest.fixture
def load_redshift_1_1():
    # csvフォーマットのテストデータをRedshiftにCOPYする処理

# テスト関数実行前にテストデータが投入される
def test_1_1(load_redshift_1_1):
    # テスト対象プログラムを実行する処理

Spectrumのケース

• テストデータをそのままのプリフィクス/パーティション構造でS3の該当locationパスにアップロードする

4. 結果検証

テスト対象プログラムの実行結果を検証する
テスト対象プログラムがRedshiftを参照(SELECT)する場合は、Redshiftに入っているデータを検証する必要はない
テスト対象プログラムがRedshiftを変更(INSERT, UPDATE, DELETE)した場合は、Redshiftの結果(現実)データとあらかじめ用意した予想データを比較検証する

Redshiftの結果データを検証する実装例

import pandas as pd

def test_1_1(load_redshift_1_1):
    # テスト対象プログラムを実行する処理

    # ex_df = csvフォーマットの予想データをpandas DataFrameで取得する処理

    # ac_df = 現実データをRedshiftからpandas DataFrameで取得する処理

    # 2つのDataFrameを比較検証する
    assert pd.testing.assert_frame_equal(
        # 主キー(id)をindexにする
        ex_df.set_index('id'),
        ac_df.set_index('id'),
        # index, columnの順序を気にしない
        check_like=True
    ) is None

Spectrumのケース

• Spectrumのテーブルを変更することはできないので、S3にあるデータオブジェクトを検証することはない

課題

• (再掲)すでにRedshiftに大量のデータが存在する場合、退避と復元に時間がかかりpytest実行がスムーズでなくなる
• 1つのデータベースを使用するためpytestの同時実行ができず、同時実行制御が必要になる

自動テスト方式別案

見送った自動テスト方式の別案

データベースコピー方式

イメージ

• 既存のデータベースを使わずに、そのデータベース内の定義を複製したデータベースを使用する
• テスト対象プログラムはpytest実行時に複製したデータベースに接続して処理する
• 既存のデータが壊れないように退避/復元する必要がない
• pytestの同時実行が可能
• 以下のコストが高そうであったため見送り
  • データベースをデータベース内の定義ごと複製する方法
  • テスト対象プログラムが接続するデータベース接続情報を複製したデータベース接続情報に差し替える方法

テスト用新規データベース作成方式

イメージ

• 既存のデータベースを使わずに、既存のデータベースと同じ定義の新規データベースを作成する
• テスト対象プログラムはpytest実行時に新規作成したデータベースに接続して処理する
• DDLは管理されているGitリポジトリ等から取得する
• 既存のデータが壊れないように退避/復元する必要がない
• pytestの同時実行が可能
• 以下のコストが高そうであったため見送り
  • 管理されたDDLを取得して新規データベースを作成する方法
  • テスト対象プログラムが接続するデータベース接続情報を新規作成したデータベース接続情報に差し替える方法