Python: motoでS3・DynamoDB・SQSのモックを作る

Python AWS

AWSで稼働するアプリケーションのユニットテストを作るときに厄介なのが、依存しているマネージドサービスのモックをどうやって整えるかです。無しというわけにはいかないですが、unittest.Mockなどで自作するのもかなり大変です。

こうしたモックを簡単に作れるようにアシストするのがmotoです。
今回は、Webやバッチなどのサービスアプリケーションから、AWSのAPIを通じてアクセスすることが多いS3・DynamoDB・SQSに絞ってmotoの使い方を紹介していきます。

moto

motoはAWSのマネージドサービスのモックを提供します。

github.com

汎用的なモックライブラリでは、テスト作成者が、テスト対象のモジュール内で呼び出しているboto3コードのクラスや返り値などを自前で書き換えることが多いかと思います。

motoでは同じboto3のインタフェースを使ってマネージドサービスをオブジェクトとして作成して、アノテーションを使ってテストコードに差し込むことで実現します。いわば仮想的なAWS環境を作成して、その上で稼働するアプリケーションとしてテストするのと同等になります。
これによって、テスト対象コードの内部実装への依存が切れます。つまり、テスト対象コードが呼び出すboto3のメソッドが変わっても、テストコードを変える必要がなくなります。モックのための実装も減り、テストコードも簡潔になります。 さらに、Flaskサーバとして独立させる機能 (Standalone Server Mode) も持っており、Pythonに限らず他の言語で書かれたアプリケーションのテストにも利用できます。

一方で、motoの欠点はサポートしているAWSマネージドサービスとAPIが限られていることです (以下で一覧化されています) 。また、マネージドサービスの作成をboto3のインタフェースで行う必要がありますので、テスト作成時にその仕様を調べる必要もあります。

https://github.com/spulec/moto/blob/master/IMPLEMENTATION_COVERAGE.md

motoを使ったユニットテスト

実際にmotoを使ったユニットテストを見ていきます。

いずれも大きく2ステップです。

  1. モックするマネージドサービスをテストメソッド (またはクラス) にアノテーションする
  2. テスト対象コードが依存しているマネージドサービスをboto3で作成する (必要に応じてデータも投入)
  3. テストコード内でテスト対象メソッドを呼び出して出力などをassert

motoとboto3をインストールします。pytestはオプションです。

$ python --version
Python 3.7.4

$ pip install pytest boto3 moto

S3

最初にS3です。テスト対象となる関数 (main.py) を以下に示します。

  • upload_to_bucket: moto-exampleバケットの"data/"配下に所定パスのファイルをアップロード
  • download_from_bucket: moto-exampleバケットの"data/"配下のファイルを所定パスへダウンロード
import boto3
import datetime

def upload_to_bucket(file_path: str, file_name: str) -> bool:
    s3_client = boto3.client("s3")

    _ = s3_client.upload_file(file_path, "moto-example", "data/" + file_name)

    return True

def download_from_bucket(file_name: str, file_path: str) -> bool:
    s3_client = boto3.client("s3")

    _ = s3_client.download_file("moto-example", "data/" + file_name, file_path)

    return True

次にテストコード (test_main.py) です。

@mock_s3アノテーションに着目してください。これによってテストクラス内でのboto3を使ったS3アクセスは、実際のS3サービスにアクセスすることなく、モックオブジェクトがリクエスト・レスポンスを代わりに返します。このモッククラスはサービス (リソース) ごとに定義されています。
また、テストコード内でboto3を介してバケットを生成しています。モックされているのでテストに必要なデータもboto3オペレーションで完結できます。

  • 例外時もboto3のインタフェース同様、botocoreの例外オブジェクト (ここではClientError) がスローされます
import pytest
import boto3
from botocore.exceptions import ClientError
from moto import mock_s3

# テスト対象となる関数
from .main import (
    upload_to_bucket, download_from_bucket,
)

@mock_s3
class TestS3Methods:
    bucket = "moto-example"

    def test_upload_succeed(self):
        # バケットの生成
        s3 = boto3.resource("s3")
        s3.create_bucket(Bucket=TestS3Methods.bucket)

        assert upload_to_bucket("./data/example.txt", "example.txt")

        # アップロードされたファイルをGet
        body = s3.Object(TestS3Methods.bucket, "data/example.txt").get()["Body"].read().decode("utf-8")

        assert body == "Hello, world!"

    def test_download_failed(self):
        s3 = boto3.resource("s3")
        s3.create_bucket(Bucket=TestS3Methods.bucket)

        # botocoreの例外クラスがスローされる
        with pytest.raises(ClientError):
            download_from_bucket("nonexist.txt", "output/example.txt")

例えば "data/" を忘れて間違ったパスへアップロードされるようになっていたとしても...

def upload_to_bucket(file_path: str, file_name: str) -> bool:
    s3_client = boto3.client("s3")

    # _ = s3_client.upload_file(file_path, "moto-example", "data/" + file_name)
   _ = s3_client.upload_file(file_path, "moto-example", file_name)

    return True

テストで気づくことができます。

FAILED test_main.py::TestS3Methods::test_upload_succeed - botocore.errorfactory.NoSuchKey: An error occurred (NoSuchKey) when calling the GetObject operation: The specified key does not exist.

DynamoDB

DynamoDBも2つの関数をテストします。moto-exampleテーブルはuser_idが数値型ハッシュキーとなります。

  • put_to_dynamo: user_id, access_count, last_accessed_atのアイテムをmoto-exampleテーブルへputする
  • get_from_dynamo: user_idをキーとしてmoto-exampleテーブルのアイテムをgetして辞書型で返す
import boto3
import datetime

def put_to_dynamo(user_id: int, access_count: int, last_accessed_at: datetime.datetime):
    dynamo_client = boto3.client("dynamodb")

    item = {
        "user_id": {"N": str(user_id)},
        "access_count": {"N": str(access_count)},
        "last_accessed_at": {"S": last_accessed_at.isoformat()},
    }

    dynamo_client.put_item(TableName="moto-example", Item=item)

def get_from_dynamo(user_id: int) -> dict:
    dynamo_client = boto3.client("dynamodb")

    key = {"user_id": {"N": str(user_id)}}

    item = dynamo_client.get_item(TableName="moto-example", Key=key)["Item"]

    return {
        "user_id": int(item["user_id"]["N"]),
        "access_count": int(item["access_count"]["N"]),
        "last_accessed_at": datetime.datetime.fromisoformat(item["last_accessed_at"]["S"]),
    }

テストコードは以下です。putとgetをまとめて1メソッドでテストしており (実用上はテストデータを作って1関数ずつテストするほうが望ましいです) 、setup_methodでテーブルを生成しています。

注意すべきはmock_dynamodb2を使っている点です。mock_dynmodbもありますが、最近のboto3では2の方を使うべきです (イシュー) 。

import datetime
import pytest
import boto3
from moto import mock_dynamodb2

# テスト対象
from .main import put_to_dynamo, get_from_dynamo

@mock_dynamodb2
class TestDynamoMethods:
    def setup_method(self, method):
        dynamodb = boto3.resource("dynamodb", region_name="ap-northeast-1")
        dynamodb.create_table(
            TableName="moto-example",
            KeySchema=[{"AttributeName": "user_id", "KeyType": "HASH"}],
            AttributeDefinitions=[{"AttributeName": "user_id", "AttributeType": "N"}],
        )

    def test_put_get_succeed(self):
        put_to_dynamo(user_id=33, access_count=10, last_accessed_at=datetime.datetime(2020, 3, 21, 10, 30, 15))

        item = get_from_dynamo(user_id=33)
        assert item["user_id"] == 33
        assert item["access_count"] == 10
        assert item["last_accessed_at"] == datetime.datetime(2020, 3, 21, 10, 30, 15)

SQS

最後にSQSです。

  • send_to_sqs: SQSへ本文がbodyとなるメッセージを1通送信する
  • receive_from_sqs: SQSからメッセージを1通受信して、本文とreceipt_handleのタプルを返す
    • メッセージが無い場合は、いずれの値もNone
import boto3

def send_to_sqs(queue_url: str, body: str):
    sqs_client = boto3.client("sqs")

    response = sqs_client.send_message(
        QueueUrl=queue_url,
        MessageBody=body
    )

    return response["MessageId"]

def receive_from_sqs(queue_url: str) -> (str, str):
    sqs_client = boto3.client("sqs")

    response = sqs_client.receive_message(QueueUrl=queue_url)

    if "Messages" not in response or len(response["Messages"]) == 0:
        return None, None

    message = response["Messages"][0]

    return message["Body"], message["ReceiptHandle"]

SQSではアカウントIDを含むURLを引数に持たせる必要があります。そのため、create_queueから返されるQueueUrlを保存し、テストしたい関数に渡しています。(通常、こうしたURLは引数で直接渡されるのではなく、設定ファイルや環境変数などでインスタンス生成時に渡ってくるものかと思います。)

import pytest
import boto3
from mock_sqs

# テスト対象
from .main import send_to_sqs, receive_from_sqs

@mock_sqs
class TestSqsMethods:
    def setup_method(self, method):
        sqs = boto3.client('sqs')
        response = sqs.create_queue(QueueName="moto-example")
        self.queue_url = response["QueueUrl"]

    def test_send_receive_succeed(self):
        assert send_to_sqs(queue_url=self.queue_url, body="Hello, world!")

        body, receipt_handle = receive_from_sqs(queue_url=self.queue_url)
        assert body == "Hello, world!"
        assert receipt_handle

    def test_receive_empty(self):
        body, receipt_handle = receive_from_sqs(queue_url=self.queue_url)
        assert body is None
        assert receipt_handle is None

まとめ

今回はmotoを使ってS3・DynamoDB・SQSに依存するアプリケーションのテストを実装してみました。

複雑になりがちなモックの作成を省力化できる点もそうですが、個人的にはテスト前にあるべき状態 (想定している状態) というのがboto3でコード化されているので、仕様の明瞭性が大きく上がるのも、motoの大きなメリットだと思います。積極的に使っていきましょう。