はじめに

AWSリソースを扱うPythonのテストコードを書く際、テスト実行のたびにリソースあるいはS3ならオブジェクトを作ったり消したりする必要があり、非効率に感じることがあります。
そこでmotoという boto3(botoやboto-core)の結果をシュミレートしてくれるモジュールを試してみたのでその使い方とサンプルコードをご紹介します。

S3やEC2のサンプルは多くありましたが、SQS/SNSに関するサンプルが少なかったため、そちらを中心に記事にしました。

`moto`とは

AWSサービスを簡単にMock化できるようにするライブラリです。
https://github.com/spulec/moto
http://docs.getmoto.org/en/latest/

準備

pipでインストール。

install

$ pip install moto

サンプルコード

対象AWSリソース

処理の流れ

SNSとSQSをmotoを用いて作成、subscriptionを設定
SNSへメッセージを送信し、publishされたSQSメッセージを取得

コード

sqs_sns.py

# -*- coding:utf-8 -*-
from pprint import pprint
from moto import mock_sns, mock_sqs
import boto3
import json

### 準備ここから ###
print("[Preparation start.]")

# mock化開始
mock_sns = mock_sns()
mock_sqs = mock_sqs()
mock_sns.start()
mock_sqs.start()

# SNS Topicを作成
client_sns = boto3.client("sns", region_name="us-east-1")
res = client_sns.create_topic(Name="some-topic")

topic_arn = res["TopicArn"]
print("Topic ARN:\n  {}".format(topic_arn))

# SQS キューを作成
client_sqs = boto3.client("sqs", region_name="us-east-1")
res = client_sqs.create_queue(
    QueueName="some-queue",
)
queue_url = res["QueueUrl"]
queue_arn = client_sqs.get_queue_attributes(
    QueueUrl=queue_url,
)["Attributes"]["QueueArn"]

print("Queue ARN:\n  {}".format(queue_arn))

# SQSへのsubscribe設定
response = client_sns.subscribe(
    TopicArn=topic_arn,
    Protocol="sqs",
    Endpoint=queue_arn
)

print("[Preparation finish.]")
### 準備ここまで ###

### mockを用いた処理ここから ###
print("[Main processing start.]")

# SNSへメッセージ送信
TEST_MESSAGE = {
    "title": "this is a test message.",
    "content": {
        "Key": "Value",
    }
}

response = client_sns.publish(
    TopicArn=topic_arn,
    Message=json.dumps(TEST_MESSAGE),
    MessageStructure="json",
)

print("Message ID:\n  {}".format(response["MessageId"]))

# SQSからメッセージ受信
response = client_sqs.receive_message(
    QueueUrl=queue_url,
)

pprint(json.loads(response["Messages"][0]["Body"])["Message"])

mock_sns.stop()
mock_sqs.stop()
print("[Main processing end.]")
### mockを用いた処理ここまで ###

処理結果（sqs_sns.py）

[Preparation start.]
Topic ARN:
  arn:aws:sns:us-east-1:123456789012:some-topic
Queue ARN:
  arn:aws:sqs:us-east-1:123456789012:some-queue
[Preparation finish.]
[Main processing start.]
Message ID:
  c96c9958-4d2c-4496-be0a-869ffff21bc9
'{"title": "this is a test message.", "content": {"Key": "Value"}}'
[Main processing end.]

アカウントID：123456789012のSNS Topic/SQSキューを作成し、その間でメッセージのpublishができることを確認できました。

補足

すこし面倒な点

開発/検品/本番環境にはコンソール/Terraform等で作成したリソースがあるかと思いますが、それと同等のリソースをテストコードの中で作成する必要があります。
複雑な構成をmock化しようとすると、そのリソース作成のコードの正しさが判断しづらくなるため、mockを用いる箇所は最小限にするのが良いのではないかと思います。

mockの有効範囲

mockを有効化するには、サンプルコードの上部にあるようにmock_sns.start()といった記述が必要になります。
mock_sns.start()/mock_sns.stop()で囲まれた範囲でのみmockが有効になり、それ以外の場所でmockを参照しようとしてもエラーとなります。

mock有効範囲外でのmock参照

# 上のサンプルコードに付け足し

response = client_sns.publish(
    TopicArn=topic_arn,
    Message=json.dumps(TEST_MESSAGE),
    MessageStructure="json",
)

結果

Traceback (most recent call last):
  File "xxx/sns_sqs.py", line 80, in <module>
    MessageStructure="json",
  File "$HOME/.pyenv/versions/3.6.0/lib/python3.6/site-packages/botocore/client.py", line 314, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "$HOME/.pyenv/versions/3.6.0/lib/python3.6/site-packages/botocore/client.py", line 612, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidClientTokenId) when calling the Publish operation: The security token included in the request is invalid.

なお、mockの有効化の方法は３種類用意されています。今回はRaw useの方式を取っています。詳細は本家のREADMEをご参照ください。
https://github.com/spulec/moto#usage
- Decorator
- Context Manager
- Raw use

moto（boto3のmockモジュール）の使い方：SQS/SNS編