RDS Data APIのテスト手法について

単体テスト AWS AWS SAM AWS Lambda Python Amazon RDS 統合テスト pytest Serverless

こんにちは。@enkw_ です。AWS LambdaとServerless Advent Calendar 2022 の 14 日目の記事です。

qiita.com

テストが遅い

とあるプロジェクトで Amazon Aurora Serverless v1 の Data API(以下 Data API)を利用した API 開発を行っていて、pytest で単体テストを書いていました。API やモジュールが増えるにつれてテストの量も増えていき、最終的に 100 を超えるテストケースが存在し、自動テストの実行に 10 分以上掛かっていました。

テスト実行に 10 分以上掛かる理由は、テストケース毎に以下の手順を踏んでいたためです。

  1. テスト実行に必要なテーブルを pytest.fixture(scope="function", autouse=True) で作成
  2. テスト実行
  3. (1) で作成したテーブルを削除

一見「こんなにかかるもの？」と思いがちですが、Data API を利用してインターネット経由でクエリを実行しているため通信コストが高く、特に Fixture setup / teardown に時間が掛かります。Docker を利用して MySQL コンテナを立ち上げられれば楽なのですが、Data API を利用している都合上 Docker でのシミュレートは難しい状況でした。

また、(1) については各テストモジュールにベタ書きしており、リポジトリ内に存在するスキーマ( schema.sql など)とは別で記載しているため、例えばスキーマに変更があった場合はテストモジュールも修正しなければならずメンテナンスコストが高い状態でした。またテストモジュールに記載したスキーマの更新を忘れ、実際のスキーマとの差異でバグが発生する問題もありました。

更にテストモジュールの行数が多く、テーブルの作成と削除、INSERT 処理なども含まれており、合わせて 1000 行を超えるテストモジュールが多数存在し、可読性が低い問題もありました。

これらの問題の一部はまだ解決できておらず、毎日自動テストが timeout している状況なのですが「こういう風に実装したらコストを低く、かつ信頼性のあるテストを書くことができるかも」という考察を書いていきたいと思います。

前提

  • API は API Gateway と Lambda(Python3.9) で構築
  • Aurora は MySQL-Compatible
  • テストは pytest で実施
  • 開発環境の Aurora に pytest 用 DB を作成

セッション毎に Fixture setup を行う

前述した通り、テストケース毎にテーブルの作成と削除を行っているためテスト時間が掛かります。 そのため、1 pytest 実行あたり 1 回のみテーブルの作成を行い、テストが終了したタイミングでテーブルを削除します。 pytest を利用する場合、conftest.py で全テストの共通処理を書けるため、以下のような Fixture を用意することでテスト開始時に DB やテーブルの作成、テスト終了時に DB を削除するといった処理が可能です。

# tests/unit/conftest.py

@pytest.fixture(scope="session", autouse=True)
def db_name():
    prefix = "test_"
    letters = "".join(choice(ascii_letters) for _ in range(10))

    return prefix + letters

@pytest.fixture(scope="session", autouse=True)
def create_schema(db_name):
    execute_statement(sql=f"CREATE DATABASE IF NOT EXISTS {db_name}")

    with open(SCHEMA, "r") as f:
        sqls = f.read().split(";")

    for sql in sqls[:-1]:
        execute_statement(sql=sql, database=db_name)

    yield

    execute_statement(sql=f"DROP DATABASE IF EXISTS {db_name}")

DB 名は db_name Fixture でランダムに生成してその名前で DB を作成、スキーマが定義されたファイルを open で読み込んで配列に格納します。最後に for で SQL 単位でテーブルを作成し、テストが終了したら DB を削除します。実際の DB に適用されているスキーマを利用しているため、先に挙げたテストモジュールと実際の DB との差異が発生する問題は解消しました。

DDL は for を使わずに一発で流せるとスマートなのですが、Data API の ExecuteStatement API で試すと BadRequestException (SQLState: 42000) が発生します。現状は 1 DDL ずつ実行する必要がありそうです。

並列でテストを実行する

並列でテストを実行することでテスト時間の短縮が見込めます。ただスレッド A のテストとスレッド B のテストが同じ DB に対してクエリを実行してしまう恐れもあるため考慮が必要です。今回の例では、先ほど定義した create_schema Fixture で、動的に DB 名を生成したため、今のところ競合は起きなさそうです。

現在 pytest-xdist を利用した並列テストを検証中で、テストケースによっては TRUNCATE の処理が必要になってくるかもしれません。その際は、以下のように TRUNCATE を実施する Fixture を用意しておくと便利そうです。

github.com

# tests/conftest.py

@pytest.fixture(scope="function", autouse=False)
def truncate(db_name):
    def _truncate(table: str):
        execute_statement(sql=f"TRUNCATE TABLE {table}", database=db_name)

    return _truncate

@pytest.fixture(scope="function", autouse=False)
def truncate_all(truncate, db_name):
    records = execute_statement(sql="SHOW TABLES", database=db_name)
    tables = [record[0].get("stringValue") for record in records["records"]]

    for table in tables:
        truncate(table)

こちらは id:mizdra さんの記事を大いに参考にさせてもらいました。ありがとうございました。

www.mizdra.net

pytest-xdist は pytest で -n オプションを指定することで、並列でのテスト実行が可能です。以下はテストの実行例(上が並列なし / 下が並列あり)で、gw0 と gw1 という worker が作成されていることが分かります。テストケースが 5 件と少ないため、worker を作成するオーバーヘッドもあってか並列なしのテストの方が実行速度が若干速いですね。テストケースが多くなればなるほど、並列テストの恩恵が受けられると思います。

pytest-xdist

統合テスト

handler の単体テストを書くことで Lambda 関数全体でのテストは可能ですが、実際の環境で API Gateway 経由のリクエストとなる場合は統合テストを実施した方が良いです。「handler の単体テストを書いたけど、デプロイしたら想定しないバグが発生した」という経験があり、やはり提供する API は HTTP での動作確認を行った方が手戻りは少ないです。handler の引数に渡す event の Request payload は自分たちで書く必要があり、どうしても typo や型の指定ミスが発生します。

統合テストを書くことで、「API が仕様通りに動作しているか」を確認できますし、より信頼性の高いテストとなるはずです。

# tests/integration/test_get_user_api.py

import requests

def test_api():
    response = requests.get(f"{BASE_URL}/user/john")

    assert response.status_code == 200
    assert response.json() == {"message": "hello john"}

自動テスト

GitHub Actions で自動テストを実施しています。以前書いた以下の記事を参考に、OIDC を利用して AssumeRole しています。

enokawa.hatenablog.jp

後述するサンプルコードには記載していませんが、最低限、以下の権限があればテストは実行できます。

# template.yaml

Policies:
  - PolicyName: !Sub "pytest-policy"
    PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action:
              - "secretsmanager:ListSecrets"
            Resource: "*"
          - Effect: Allow
            Action:
              - "secretsmanager:DescribeSecret"
              - "secretsmanager:GetSecretValue"
            Resource: !Sub "arn:aws:secretsmanager:{AWS::Region}:{AWS::AccountId}:secret:<secret-name>"
          - Effect: Allow
            Action:
              - "rds-data:ExecuteStatement"
            Resource: !Sub "arn:aws:rds:{AWS::Region}:{AWS::AccountId}:cluster:<cluster-name>"

CI 用に DB のユーザと、Secrets を作っておいて、特定の DB のみ操作可能な権限を GRANT すると安心ですね。

CREATE USER 'pytest'@'%' identified by 'password';
GRANT ALL ON `test_%`.* to 'pytest'@'%';

おわりに

t_wada さんのこのツイートを見て、ハッとさせられました。「Data API のテスト遅いし、もう mock でいいのでは..」と思っていた時に目にして、このままではいけないなと。自作自演のテストを書いて信頼性のないテストを書いて満足するよりも、テスト時間もそこそこ速くして信頼性を上げるテストを書くように思考した方がカッコいいよなと思いこの記事を書きました。

少しタイトルの趣旨とズレた記事となってしまいましたが、この記事で「いやこれは違う」とか「このツール使えば解決では？」みたいなものがあれば教えてもらえると喜びます。

サンプルコード

github.com

参考

ありがとうございました！

www.mizdra.net

devblog.thebase.in

github.com

zenn.dev