moto（boto3のmockモジュール）の使い方：SQS/SNS編
はじめに
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でインストール。
$ pip install moto
サンプルコード
対象AWSリソース
- SQS
- SNS
処理の流れ
- 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.]
motoを用いて作成、subscriptionを設定# -*- 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を用いた処理ここまで ###
[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を参照しようとしてもエラーとなります。
# 上のサンプルコードに付け足し
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
Author And Source
この問題について(moto（boto3のmockモジュール）の使い方：SQS/SNS編), 我々は、より多くの情報をここで見つけました https://qiita.com/ck_fm0211/items/08a7bc5a0c98de112cb7著者帰属：元の著者の情報は、元のURLに含まれています。著作権は原作者に属する。
Content is automatically searched and collected through network algorithms . If there is a violation . Please contact us . We will adjust (correct author information ,or delete content ) as soon as possible .