How can I test SMS using AWS SNS (boto3/moto) in Python/Django?

user_686617 0
Viewed 2052

I'm working on an API that sends SMS in some endpoints using AWS SNS. In tests that touch these endpoints, the mock_sns decorator is used. However, no assertions are made regarding the number or content of the SMS messages sent. I suppose this is only there to avoid errors. In my case, however, I actually need to check that one SMS message was sent and that it contains a specific text. How can I do that? How should my test look like?

I know how to do it using plain Python mocking but since the project already has moto as a dependency I figured I'd give it a try, but I'm surprised I couldn't find any example of such a simple use case. Or perhaps I'm misunderstanding the purpose of this library?

@mock_sns
def test_my_endpoint():
    response = client.post("/my/endpoint/", ...)
    # assert one SMS was sent???
    # assert "FooBar" in SMS???
moto boto3 pytest testing python
2 Answers

Since you want to use the moto library, why not use sqs to test that only one notification has been sent using sns?

The sns + sqs connection, sometimes called as fanout, will achieve the specified test as the notification will be stored in sqs and you can check it right after the call. Check the test_fanout method below.

On the other side and with the information provided, I would use botocore stubs or python mocks to check that the call is already made the way you want. In case of using stubber, check the test_stubber_sms method below.

The code assumes moto >= 1.3.14, boto3 1.18, botocore 1.21.0 and 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()
user_43111600

Unfortunately, there doesn't seem to be a direct way to do this as there is no such API call in the boto3 library.
What you can do, is to create a subscriber (mock), register to the sns and assert on the subscriber. Examples of such tests can be found in the internal tests of moto iteslf:

@mock_sns
def test_publish_sms():
    client = boto3.client("sns", region_name="us-east-1")

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

    result.should.contain("MessageId")
    if not settings.TEST_SERVER_MODE:
        sns_backend = sns_backends["us-east-1"]
        sns_backend.sms_messages.should.have.key(result["MessageId"]).being.equal(
            ("+15551234567", "my message")
        )

From: https://github.com/spulec/moto/blob/master/tests/test_sns/test_publishing_boto3.py

user_10492340
Related