AWS x Pythonで自動テストを書いていくあなたに

こんにちわ alivelimb です。
Pythonista のみなさん、自動テスト書いてますか？書き捨てでないコードを書くのであれば、自動テストを書きましょう。本記事では AWS x Python で自動テストを書く際に役立つLocalStackとmotoを紹介します。

はじめに

本記事では テストフレームワークとしてpytestを利用します。また以下に示す「AWS S3 に置いてある画像をnumpy.ndarray形式で取得する」関数をテスト対象の具体例として考えていきます。

import boto3
import cv2
import numpy as np
from mypy_boto3_s3.service_resource import Bucket

def get_image_ndarray_from_s3(bucket: Bucket, key: str) -> np.ndarray:
    image_bytes = bucket.Object(key).get()["Body"].read()
    image_ndarray = cv2.imdecode(np.asarray(bytearray(image_bytes)), cv2.IMREAD_COLOR)
    return image_ndarray


if __name__ == "__main__":
    bucket_name = "test-bucket"
    key = "test_image.png"

    s3 = boto3.resource("s3")
    bucket = s3.Bucket(bucket_name)
    img = get_image_ndarray_from_s3(bucket, key)

この関数について 2 点補足します。

まず 画像の読み込みにはOpenCVを使っています。Pillowを使ってもよいと思いますが、普段の業務で使っている方をチョイスしました。

次に TypeHint です。私は基本的に TypeHint を書くようにしています。ただ、執筆時点で boto3 の TypeHint は公式ではサポートされていないようでしたので、mypy_boto3_builderを別途インストールして活用しています。

動作環境について

本記事で検証を行った環境・各種バージョンは以下の通りです

• macOS Monterey v12.3.1
• Docker version 20.10.12
• docker-compose version 1.29.2
• Python 3.8.7
  • boto3 = "^1.21.21"
  • numpy = "^1.22.3"
  • opencv-python = "^4.5.5"
  • boto3-stubs = {extras = ["s3"], version = "^1.22.8"}
  • moto = {extras = ["s3"], version = "^3.1.8"}

普通にテストしちゃだめなの？

この関数の自動テストを書こうと思ったとき、普通にテストするとしたら以下のようになるでしょうか。

1. テスト用の S3 バケットを作成する
2. 作成した S3 バケットにテストデータ(画像をアップロードする)
3. 以下のようなテストコードを書いてテストする

def test_get_image_ndarray_from_s3() -> None:
    test_bucket_name = "作成したバケット名"

    s3 = boto3.resource("s3")
    test_bucket = s3.Bucket(test_bucket_name)
    expected = cv2.imread("ローカルのテスト画像パス")
    actual = get_image_ndarray_from_s3(test_bucket, "アップロード先のS3パス(key)")
    np.testing.assert_array_equal(expected, actual)

これでもテストは問題なく動きますが、以下のようなデメリットがあります。

1. S3 との通信が必須のため、その分テストに時間がかかる
2. テスト用の S3 のバケット作成や削除も自動化する場合、さらに時間がかかる
3. テスト用の S3 バケットを他の開発者(テスター)と共有する場合、他の開発者がバケット内の状態を変更する恐れもある

これから紹介する LocalStack や moto はこれらの問題を解決してくれます。

LocalStack

まずLocalStackです。LocakStack はローカルマシン内にコンテナとして立ち上げることで、擬似的な AWS 環境として動作します。コンテナのため Docker 環境が必要になりますが、そこがクリアできればローカル環境だけでテストを完結させることが可能です。

LocalStack には無料版と有料版があり、RDS, Athena など有料版でしか使えない AWS サービスもあります。自分の使いたいサービスが無料版で使えるかは公式ドキュメントで確認できます。

また今回は Python で利用していますが、LocalStack は Python でなくても利用可能です。LocalStack のコンテナを立ち上げてしまえば、AWS CLI や他の言語からアクセスすることもできます。

動作環境の準備

LocalStack を使うにはDockerをインストールしておく必要があります。また、本記事ではDocker Composeを用いて LocalStack 環境を構築するのでこちらもインストールしておく必要があります。LocalStack CLIを用いる場合はpipでインストール出来るようですが、こちらは今回利用しません。

Docker, Docker Compose の環境構築については様々な記事で紹介されているので、本記事では割愛します。私のおすすめは公式ドキュメントの日本語訳です。

Docker, Docker Compose のインストールが出来たらdocker-compose.ymlを作成します。今回は公式リポジトリの docker-compose.ymlを一部修正したものを用意しました(Pro 版でのみ必要な記述を削除など)。

version: "3.8"

services:
  localstack:
    container_name: "${LOCALSTACK_DOCKER_NAME-localstack}"
    image: localstack/localstack
    network_mode: bridge
    ports:
      - "127.0.0.1:4510-4559:4510-4559" # external service port range
      - "127.0.0.1:4566:4566" # LocalStack Edge Proxy
    environment:
      - DEBUG=${DEBUG-}
      - DATA_DIR=${DATA_DIR-}
      - LAMBDA_EXECUTOR=${LAMBDA_EXECUTOR-}
      - HOST_TMP_FOLDER=${TMPDIR:-/tmp/}localstack
      - DOCKER_HOST=unix:///var/run/docker.sock
    volumes:
      - "${TMPDIR:-/tmp}/localstack:/tmp/localstack"
      - "/var/run/docker.sock:/var/run/docker.sock"

あとは以下のコマンドで LocalStack のコンテナを立ち上げれば準備完了です。

docker-compose up -d

LocalStack を用いたテスト

今回の設定ではhttp://localhost:4566で LocalStack コンテナにアクセス可能です。AWS CLI や boto3 などの AWS SDK のendpoint urlにこの URL を渡せば実際の AWS サービスではなく、LocalStack コンテナにアクセスしてくれます。

AWS CLI

aws s3 ls --endpoint-url="http://localhost:4566"

aws configureや環境変数でリージョンを指定していない場合、--regionでリージョンを指定する必要があります。設定の優先順位は公式ドキュメントを参照して下さい。

では LocalStack を使ってテストを書いてみましょう。

def test_get_image_ndarray_from_s3() -> None:
    bucket_name = "test-bucket"
    region_name = "ap-northeast-1"

    # テストバケットの準備
    s3 = boto3.resource("s3", endpoint_url="http://localhost:4566", region_name=region_name)
    test_bucket = s3.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": region_name})
    test_bucket.upload_file("tests/data/test_image.png", "test_image.png")

    expected = cv2.imread("tests/data/test_image.png")
    actual = get_image_ndarray_from_s3(test_bucket, "test_image.png")
    np.testing.assert_array_equal(expected, actual)

AWS CLI と同様、endpoint urlを指定することで LocalStack にアクセスすることができます。これで実際の S3 バケットがなくても、AWS アカウントを持ってすらいなくても単体テストを書くことが出来ます。

小話: LocationConstraintについて

先ほどのテストコードでLocationConstraintが気になった方もいるのではないでしょうか。実は boto3 で S3 バケットをus-east-1以外に作成する場合に指定する必要があります。boto3 のドキュメントには

Specifies the Region where the bucket will be created. If you don't specify a Region, the bucket is created in the US East (N. Virginia) Region (us-east-1).

との記述があります。ざっくり訳すと「バケットを作成するリージョンを指定するよ。指定しないとus-east-1に作成されるよ」でしょうか。LocationConstraintを指定しないと以下のようなエラーが表示されます。

botocore.exceptions.ClientError: An error occurred (IllegalLocationConstraintException) when calling the CreateBucket operation: The unspecified location constraint is incompatible for the region specific endpoint this request was sent to.

「いやいや AWS CLL でaws s3 mbする時はそんなことなかったけど？」と思いましたが、AWS 公式ブログに答えがありました。どうやら AWS CLI で S3 を操作するのはs3だけでなくs3apiもあるようです。試しに以下のコマンドで検証してみたら同様のエラーが発生しました。

aws s3api create-bucket --bucket=test-bucket --region=ap-northeast-1 --endpoint-url=http://localhost:4566

# An error occurred (IllegalLocationConstraintException) when calling the CreateBucket operation: The unspecified location constraint is incompatible for the region specific endpoint this request was sent to.

moto

次にmotoです。moto は boto3 のモックを提供してくれます。LocalStack と同様に実際の AWS リソースがなくても、AWS アカウントを持ってすらいなくてもテストを書くことができます。ただし boto3 のモックなので、LocalStack とは異なり Python でしか利用できません。

非常に多くの AWS リソースのモックを提供してくれていますが、全ての boto3 メソッドをモック化しているわけでないので、moto の公式ドキュメントで使いたいメソッドに対応しているか確認して下さい。

モックとは何か

「そもそもモックって何？」という方もいるでしょう。簡単に言うと「あらかじめ決められた値を返すだけのオブジェクト」でしょうか。「S3 においてある大容量の CSV データを処理する関数」のテストを例に上げて考えてみます。

「処理が機能として正しいか」のみを確認したいのであれば、テストも大容量データで行う必要はありません。そこで「S3 から CSV を取得するオブジェクト」を「事前に(メモリ上に)作成した CSV 相当の値を返すオブジェクト」に置き換えれば、S3 に通信する必要も、テストのたびに重たい処理をする必要もありません。この代わりのオブジェクトがモックです。

AWS Japan の福井さんが非常に分かりやすいスライドを作成されているので、詳しくはサーバーレス開発環境とテストを参照して下さい。

モックはpytest-mockなどを使って自作することも可能ですが、moto を使えば自分で作ることなく様々なサービスをモックを利用することができます。

moto を用いたテスト

早速テストコードを見ていきましょう。

@mock_s3
def test_get_image_ndarray_from_s3() -> None:
    bucket_name = "test-bucket"
    region_name = "ap-northeast-1"

    # テストバケットの準備
    s3 = boto3.resource("s3", region_name=region_name)
    test_bucket = s3.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": region_name})

    # サンプル画像のアップロード
    test_bucket.upload_file("tests/data/test_image.png", "test_image.png")

    expected = cv2.imread("tests/data/test_image.png")
    actual = get_image_ndarray_from_s3(test_bucket, "test_image.png")
    np.testing.assert_array_equal(expected, actual)

LocalStack の時とほとんど同じですが、@mock_s3デコレータがついてるのとendpoint_urlがなくなりました。moto はデコレータや with ブロックでモック化する範囲を指定できます(withの例は後述します)。今回はデコレータなのでtest_get_image_ndarray_from_s3関数内の S3 操作はモック化されることになります。

どうやって使い分けるか

さて LocalStack と moto の 2 つを紹介してきましたが、これをどうやって使い分ければ良いのでしょうか。無料で使うことを考えると LocalStack と moto3 で使えるサービスが変わってくるので一概には言えませんが、テストコード、テストピラミッド、TestDoubleの区分け、の 3 つの観点で考えてみます。

テストコード

これまでに紹介した pytest のコードを見て「なんでfixture使わないの？」と思っていた方もいるかもしれません。fixture と moto を使ってテストを書くと以下のようになります

@fixture
def test_bucket(bucket_name: str = "test-bucket", region_name: str = "ap-northeast-1") -> Generator[Bucket, None, None]:
    with mock_s3():
        s3 = boto3.resource("s3", region_name=region_name)
        bucket = s3.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": region_name})

        # サンプル画像のアップロード
        bucket.upload_file("tests/data/test_image.png", "test_image.png")

        yield bucket


def test_get_image_ndarray_from_s3(test_bucket: Bucket) -> None:
    expected = cv2.imread("tests/data/test_image.png")
    actual = get_image_ndarray_from_s3(test_bucket, "test_image.png")
    np.testing.assert_array_equal(expected, actual)

fixture で moto を使う時はデコレータではなく、withブロックを使った方が書きやすいと思います。moto を使う場合、Python で完結するので特に後処理は必要ありません。withブロックを抜ければモック化されませんし、読み取り専用のテストなどで使いまわしたい場合はfixture のスコープをうまく使えば OK です。

一方で LocalStack はコンテナ内に状態が保持されているため、Python で自動化したければ後処理(teardown)を書く必要があります。そのため短いサイクルで単体テストを回したいのであれば moto を使うのがよいと思いますし、連結テストで Python だけで完結しない場合などは LocalStack が適すると思います。

テストピラミッド

テストピラミッドに関してはThe Practical Test Pyramidや和田さんのTestable Lambda ｜ AWS Summit Tokyo 2017(動画), Testable Lambda: Working Effectively with Legacy Lambda(動画内のスライド)などで紹介されているので詳細は割愛します。

和田さんのスライドでも紹介されているGoogle のテストサイズの観点で言うと、moto が small サイズ、LocalStack が medium サイズということになるかと思います。テスト戦略でこのような区分けがあるのであれば、各フェーズに応じて使い分けができそうです。

TestDouble の区分け

既に紹介したサーバーレス開発環境とテストでは moto をモック、DynamoDB Localなどを Fake と紹介されていました。

※DynamoDB LocalやCodeBuild Local Buildsは使ったことがないのですが、機能としては LocalStack と非常に似ているようです。なお、こちらの記事では LocalStack の DynamoDB と DynamoDB Local のベンチマークを取っており、DynamoDB Local の方がパフォーマンスが良かったそうです。

LocalStack は様々なサービスを使えるのが魅力ですが、パフォーマンス面では DynamoDB Local の方がよさそうです。DynamoDB Local の S3 版に相当する公式が出しているものは見つけられませんでしたが、こちらはMinIOなどが有名です。

まとめ

本記事では AWS x Python で自動テストを書いていくあなたに向けて、LocalStack と moto を紹介しました。LocalStack や本記事で紹介した DynamoDB Local, MinIO などは Python でなくても使えるので是非使ってみて下さい。