如何在 Python/Django 中使用 AWS SNS (boto3/moto) 测试 SMS？答案

【问题标题】：How can I test SMS using AWS SNS (boto3/moto) in Python/Django?如何在 Python/Django 中使用 AWS SNS (boto3/moto) 测试 SMS？
【发布时间】：2020-04-07 12:53:43
【问题描述】：

我正在开发一个使用 AWS SNS 在某些端点发送 SMS 的 API。在涉及这些端点的测试中，使用了 mock_sns 装饰器。但是，没有就发送的 SMS 消息的数量或内容作出任何断言。我想这只是为了避免错误。然而，就我而言，我实际上需要检查是否发送了一条 SMS 消息并且它是否包含特定文本。我怎样才能做到这一点？我的测试应该是什么样子？

我知道如何使用纯 Python 模拟来做到这一点，但由于该项目已经将 moto 作为依赖项，我想我会尝试一下，但我很惊讶我找不到任何这样的例子简单的用例。还是我误解了这个库的目的？

@mock_sns
def test_my_endpoint():
    response = client.post("/my/endpoint/", ...)
    # assert one SMS was sent???
    # assert "FooBar" in SMS???

【问题讨论】：

标签： python testing pytest boto3 moto

【解决方案1】：

既然要使用moto库，为什么不使用sqs来测试使用sns只发送了一个通知呢？

sns + sqs 连接，有时称为fanout，将实现指定的测试，因为通知将存储在 sqs 中，您可以在调用后立即查看。检查下面的 test_fanout 方法。

另一方面，根据提供的信息，我将使用botocore stubs 或 python 模拟来检查调用是否已按照您想要的方式进行。如果使用 stubber，请检查下面的 test_stubber_sms 方法。

代码假定 moto >= 1.3.14、boto3 1.18、botocore 1.21.0 和 python 3.6.9：

from moto import mock_sns, mock_sqs
import unittest
import boto3
import json
from botocore.stub import Stubber


class TestBoto3(unittest.TestCase):

    @mock_sns
    @mock_sqs
    def test_fanout(self):
        region_name = "eu-west-1"
        topic_name = "test_fanout"
        sns, sqs = boto3.client("sns", region_name=region_name), boto3.client("sqs", region_name=region_name)

        topic_arn = sns.create_topic(Name=topic_name)["TopicArn"]
        sqs_url = sqs.create_queue(QueueName='test')["QueueUrl"]
        sqs_arn = sqs.get_queue_attributes(QueueUrl=sqs_url)["Attributes"]["QueueArn"]

        sns.subscribe(TopicArn=topic_arn, Protocol='sqs', Endpoint=sqs_arn)
        sns.subscribe(TopicArn=topic_arn, Protocol='sms', Endpoint="+12223334444")

        sns.publish(TopicArn=topic_arn, Message="test")
        sns.publish(PhoneNumber="+12223334444", Message="sms test")

        message = sqs.receive_message(QueueUrl=sqs_url, MaxNumberOfMessages=10)["Messages"]
        sqs.delete_message(QueueUrl=sqs_url, ReceiptHandle=message[0]["ReceiptHandle"])

        self.assertEqual(len(message), 2)
        self.assertEqual(json.loads(message[0]["Body"])["Message"], 'test')
        self.assertEqual(json.loads(message[1]["Body"])["Message"], 'sms test')

    def test_stubber_sms(self):
        sns = boto3.client("sns")
        stubber = Stubber(sns)

        stubber.add_response(method='publish', service_response={"MessageId": '1'}, expected_params={'PhoneNumber':"+12223334444", 'Message':"my message"})
        with stubber:
            response = sns.publish(PhoneNumber="+12223334444", Message="my message")
            # Or use your method and pass boto3 sns client as argument
            self.assertEqual(response, {"MessageId": '1'})
            stubber.assert_no_pending_responses()

if __name__ == '__main__':
    unittest.main()

【讨论】：

【解决方案2】：

不幸的是，似乎没有直接的方法可以做到这一点，因为boto3 库中没有这样的 API 调用。
您可以做的是创建一个订阅者（模拟），注册到 sns 并在订阅者上断言。此类测试的示例可以在motoiteslf的内部测试中找到：

@mock_sns
def test_publish_sms():
    client = boto3.client("sns", region_name="us-east-1")
    client.create_topic(Name="some-topic")
    resp = client.create_topic(Name="some-topic")
    arn = resp["TopicArn"]

    client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="+15551234567")

    result = client.publish(PhoneNumber="+15551234567", Message="my message")
    result.should.contain("MessageId")

发件人：https://github.com/spulec/moto/blob/master/tests/test_sns/test_publishing_boto3.py

【讨论】：