How to Test AWS SNS Topics

### The Problem

Recently I had a client that I worked with who uses AWS SNS Topics to deliver notifications when an event has occurred. The application itself is written in Python and is a Flask app. Testing was core value and the project team stived to have our code coverage as high as possible. They are using many AWS features and most of the others have been relatively easy to test, however SNS topics were different.

What you'll learn

By the end of this post you'll learn how to test AWS SNS topics to ensure that your code is processing the data correctly and that you are successfully handling any errors.

Prerequisites

You will need the following:

Python3.7+ installed on your development computer.

An active AWS account that you can access the AWS access key id and secret access key. If you don't have an AWS account setup already please see my guide here.

Getting Started

I'm currently using Python 3.7, so if you are using 3.7 or higher you should have no trouble. Change directory (cd) into your local development directory:


cd ~/Dev

Next we'll clone the inital demo project:


git clone https://github.com/marty331/aws-sns-test

After you're done, cd into the project directory and then we'll create a virtual environment to work in and we'll activate that virtual environment. We do this so that the packages we install will be installed in this workspace only.


python -m venv env

source env/bin/activate

Next we'll install the following packages:


pip install -r requirements.txt

Project Setup

Setup an AWS SNS Topic and Subscription for testing, if you don't have one setup yet then follow my guide on how to do this here: How to Setup an AWS SNS Topic and Subscription

Now that we have our project directory, virtual environment and initial packages setup, let's run the project, first we'll set our FLASK_APP, FLASK_ENV, and TOPIC_ARN_KEY environment variables, then we'll run the project.


$ export FLASK_APP=awssns
$ export FLASK_ENV=development
$ export TOPIC_ARN_KEY=<your-topic-arn-key>
$ export AWS_REGION=us-west-1
$ flask run

once this is complete, you should see some output like this:


* Serving Flask app "awssns"
* Environment: dev
* Debug mode: off
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

and of course you'll want to visit the url in your browser, which should return back - "Hello, World!"

Now we can run a simple curl command to test out our SNS topic:


curl -X POST http://localhost:5000/test-sns?message=Hello%20SNS

Now, assuming you followed my guide linked above, you should get an email that looks something similar to this:

If you received the email successfully, great! Pat yourself on the back. You have sucessfully created an AWS SNS Topic and recieved a notification from it. You have proved it works, so blog post over right? Not so fast! Now the fun (for you) begins, we will test the SNS Topic with Pytest.

Messages.py Review

First, let's review the actual SNS Topic code so we can know what we need to test. The top section is all of the imports that we need. I'll skip over the standard library imports and go straight to boto3 - the main library for connecting to AWS in Python. If this were app I were optimizing for production I would import only the client from the boto3 library. Next we import the ClientError from botocore.exceptions. As you probably guessed, this is the error class associated with the boto3 library. And finally we import CFG from the cfg.py file. cfg.py is an building an implementation of the decouple library and gives up an easy way to reference environment variables from .env files or from environment variables. It also works great if you are deploying an AWS Lambda function.


# messages.py file
# imports start
import os
import json
import logging

from typing import Dict, Any
from abc import ABC

import boto3

from botocore.exceptions import ClientError

from awssns.cfg import CFG
#imports end

Setup Logging

We initialize our logger, keeping it very simple here and not adding in any configuration as we just want to get simple messages and feedback.


#initialize logger
logger = logging.getLogger()

Abstract Route

For good measure we'll create a class that extends the ABC class. We're only using this to initialize our payload but if you have multiple classes I reccomend creating a base class like this to inheirt from.



# create abstract class from ABC class, not 100% but good practice
class AbstractRoute(ABC):
    def __init__(self, payload) -> None:
        self.payload = payload

Message Class

Finally we create our Message class, which will contain all of the functionality for creating the SNS Topic and sending it to AWS. It inheirts the AbstractRoute class and therefore must contain a payload (our message). We create a function (route) that will control the flow and publishing of the SNS Topic message.



# create message class, inherit from AbscractRoute
class Message(AbstractRoute):
    # function that sends the SNS message
    def route(self) -> Dict[str, Any]:
        sns_client = self.get_sns_client()
        response = self.publish_sns_message(
            sns_client=sns_client, topic_arn=CFG.TOPIC_ARN_KEY, message=self.payload
        )
        if response["ResponseMetadata"]["HTTPStatusCode"] == 200:
            return response
        return {"ResponseMetadata": {"HTTPStatusCode": 500}}

A boto3 client is created for SNS Topics, this requires the service name and region name. The service name is "sns" and the region name will be something like "us-west-1". Assuming you have your AWS credentials setup correctly this this function will return back a boto3 SNS client.


# connect to AWS via boto3 library
    def get_sns_client(self) -> boto3.client:
        return boto3.client("sns", region_name=CFG.AWS_REGION)


# publish the message to the SNS topic
    def publish_sns_message(
        self, sns_client: boto3.client, topic_arn: str, message: Dict[str, Any]
    ) -> Dict[str, Any]:
        try:
            response = sns_client.publish(
                TopicArn=topic_arn,
                Message=json.dumps(
                    {"default": message}
                ),  # json.dumps is required by FxA
                MessageStructure="json",
            )
            return response
        except ClientError as e:
            logger.error("SNS error", error=e)
            raise e