SNS mocking with moto is not working correctly

Question

In my unit test:

def test_my_function_that_publishes_to_sns():
    conn = boto3.client("sns", region_name="us-east-1")
    mock_topic = conn.create_topic(Name="mock-topic")
    topic_arn = mock_topic.get("TopicArn")

    os.environ["SNS_TOPIC"] = topic_arn

    # call my_function
    my_module.my_method()

The the function being tested

# inside my_module, my_function...
sns_client.publish(
            TopicArn=os.environ["SNS_TOPIC"], Message="my message",
        )

I get the error: botocore.errorfactory.NotFoundException: An error occurred (NotFound) when calling the Publish operation: Endpoint with arn arn:aws:sns:us-east-1:123456789012:mock-topic not found

Doesn't make sense, that's the topic moto is suppose to have created and mocked. Why is it saying it doesn't exist? If I call conn.publish(TopicArn=topic_arn, Message="sdfsdsdf") inside of the unit test itself it seems to mock it, but it doesn't mock it for my_module.my_method() which the unit test executes. Maybe it's destroying the mocked topic too soon?

EDIT I tried this every which way and I get the exact same error:

# Using context manager
def test_my_function_that_publishes_to_sns():
    with mock_sns():
        conn = boto3.client("sns", region_name="us-east-1")
        mock_topic = conn.create_topic(Name="mocktopic")
        topic_arn = mock_topic.get("TopicArn")
    
        os.environ["SNS_TOPIC"] = topic_arn
    
        # call my_function
        my_module.my_method()


# Using decorator
@mock_sns
def test_my_function_that_publishes_to_sns():
    conn = boto3.client("sns", region_name="us-east-1")
    mock_topic = conn.create_topic(Name="mocktopic")
    topic_arn = mock_topic.get("TopicArn")

    os.environ["SNS_TOPIC"] = topic_arn

    # call my_function
    my_module.my_method()


# Using decorator and context manager
@mock_sns
def test_my_function_that_publishes_to_sns():
    with mock_sns():
        conn = boto3.client("sns", region_name="us-east-1")
        mock_topic = conn.create_topic(Name="mocktopic")
        topic_arn = mock_topic.get("TopicArn")
    
        os.environ["SNS_TOPIC"] = topic_arn
    
        # call my_function
        my_module.my_method()

Opened GitHub issue as well: https://github.com/spulec/moto/issues/3027

Answer 1

The solution for me was to have matching account ids:

def sns_client_mock():
    with moto.mock_aws():
        sns_client_mock = boto3.client('sns', region_name='eu-central-1')
        topic = sns_client_mock.create_topic(Name='abc')
        sns_client_mock.add_permission(
            TopicArn=topic['TopicArn'],
            Label="test",
            AWSAccountId=[topic['TopicArn'].split(':')[4]],
            ActionName=["Publish"],
        )
        sns_client_mock.publish(TopicArn=topic['TopicArn'], Message='test')
        yield sns_client_mock

Answer 2

Sample code below. I hope it helps somebody. The suggested fix about setting the Region was not my issue. If you are still stuck, this video is great.

Approach:

1. Create a mocked Boto3 Resource ( not a Boto3 Client ).
2. Set mock SNS Topic ARN in this new resource.
3. Overwrite the SNS Topic ARN environment var for the test.
4. Get a Boto3 Client that calls Publish to the mocked SNS Topic ARN.

I hit the below error because I set the Topic ARN to mock_topic and not arn:aws:sns:eu-west-1:123456789012:mock_topic:

botocore.errorfactory.NotFoundException: An error occurred (NotFound) when calling the Publish operation: Endpoint does not exist """

import main
import boto3
import pytest
import botocore
from moto import mock_sns

# http://docs.getmoto.org/en/latest/docs/getting_started.html
#####################################################################
# test_main.py
#####################################################################

@pytest.fixture()
def mock_message():
    return {
        "foo": "1st wonderful message.",
        "bar": "2nd wonderful message.",
        "baz": "3rd wonderful message.",
    }


@pytest.fixture()
def mock_sns_client():
    return sns_publish.get_sns_client()


def test_get_mocked_sns_client(mock_sns_client):
    assert isinstance(mock_sns_client, botocore.client.BaseClient)


mock_topic_name = "mock_topic"


@mock_sns
def test_mock_send_sns(mock_message, monkeypatch, mock_sns_client):
    """
    1. Create a mocked Boto3 Resource ( not a Boto3 Client ).
    2. Set mock SNS Topic ARN in this new resource.
    3. Overwrite the SNS Topic ARN environment var for the test.
    """
    sns_resource = boto3.resource(
        "sns",
        region_name=os.environ.get("AWS_REGION")
    )

    topic = sns_resource.create_topic(
        Name=mock_topic_name
    )
    assert mock_topic_name in topic.arn
    monkeypatch.setenv('SNS_TOPIC_ARN', topic.arn)
    assert os.environ.get("SNS_TOPIC_ARN") == topic.arn

    response = sns_publish.send_sns(mock_sns_client, mock_message)

    assert isinstance(response, dict)
    message_id = response.get("MessageId", None)
    assert isinstance(message_id, str)


#####################################################################
# main.py
# split the get Client and Publish for simpler testing
#####################################################################

import boto3
import json
import botocore
import os
from conf.base_logger import logger


# split the get Client and Publish for simpler testing
def get_sns_client():
    return boto3.client("sns", region_name=os.environ.get("AWS_REGION"))


def send_sns(sns_client, message: dict) -> dict:
    if not isinstance(message, dict):
        logger.info("message to send Slack is not in expected format")
        return None
    if not isinstance(sns_client, botocore.client.BaseClient):
        logger.info("something wrong with the SNS client")
        return None
    return sns_client.publish(
        TargetArn=os.environ.get("SNS_TOPIC_ARN"),
        Message=json.dumps({'default': json.dumps(message, indent=4, sort_keys=True)}),
        Subject='Foo\'s stats',
        MessageStructure='json'
    )

Answer 3

maybe it will help you
keep all modules in a single class and put a decorator @mock_sns on the class too for mocking the sns, also put decorator @mock_sns on the function where you are initializing you connection to sns. Example:

@mock_sns
class TestSnsMock(unittest.TestCase):

    @classmethod
    @mock_sns
    def setUpClass(cls):
        cls.conn = boto3.client("sns", region_name="us-east-1")
        cls.conn.create_topic(Name="some-topic")
        cls.response = cls.conn.list_topics()
        cls.topic_arn = cls.response["Topics"][0]["TopicArn"]

    def test_publish_sns(self):
        message = "here is same message"
        self.sns_client.publish(TopicArn=self.topic_arn, Message=message)


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

Answer 4

issue was my_module.my_method() wasn't setting a region just doing client = boto3.client("sns")

It could not find it because it was defaulting to a diff region than us-east-1 which was hard coded into the unit test