使用 moto 的 SNS 模拟无法正常工作

Question

在我的单元测试中：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()正在测试的函数# inside my_module, my_function...sns_client.publish(            TopicArn=os.environ["SNS_TOPIC"], Message="my message",        )我收到错误：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没有意义，这就是 moto 应该创建和嘲笑的主题。为什么说它不存在？如果我 conn.publish(TopicArn=topic_arn, Message="sdfsdsdf")在单元测试本身内部调用它似乎是在模拟它，但它不会为单元测试执行的 my_module.my_method() 模拟它。也许它过早地破坏了嘲笑的话题？编辑我以各种方式尝试了这个，但我得到了完全相同的错误：# Using context managerdef 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_snsdef 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_snsdef 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()也打开了 GitHub 问题：https ://github.com/spulec/moto/issues/3027

Answer 1

也许它会帮助您

将所有模块放在一个类中，并在类上放置一个装饰器@mock_sns 来模拟 sns，还将装饰器@mock_sns 放在您初始化 sns 连接的函数上。例子：

@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 2

问题是 my_module.my_method() 不是设置一个区域只是做client = boto3.client("sns")

它找不到它，因为它默认为与硬编码到单元测试中的 us-east-1 不同的区域

Answer 3

下面的示例代码。我希望它能帮助别人。关于设置区域的建议修复不是我的问题。如果您仍然卡住，这个视频很棒。

方法：

1. 创建一个模拟的 Boto3 资源（不是 Boto3 客户端）。

2. 在此新资源中设置模拟 SNS 主题 ARN。

3. 覆盖用于测试的 SNS 主题 ARN 环境变量。

4. 获取一个调用 Publish 到模拟 SNS 主题 ARN 的 Boto3 客户端。

我遇到了以下错误，因为我将主题 ARN 设置为mock_topic而不是arn:aws:sns:eu-west-1:123456789012:mock_topic：

botocore.errorfactory.NotFoundException：调用发布操作时发生错误（NotFound）：端点不存在“”“

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'

    )