Help us understand the problem. What is going on with this article?

moto(boto3のmockモジュール)の使い方:SQS/SNS編

More than 1 year has passed since last update.

はじめに

AWSリソースを扱うPythonのテストコードを書く際、テスト実行のたびにリソースあるいはS3ならオブジェクトを作ったり消したりする必要があり、非効率に感じることがあります。
そこでmotoという boto3(botoboto-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リソース

  • 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.]

アカウント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の有効化の方法は3種類用意されています。今回はRaw useの方式を取っています。詳細は本家のREADMEをご参照ください。
https://github.com/spulec/moto#usage
- Decorator
- Context Manager
- Raw use

Why do not you register as a user and use Qiita more conveniently?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away