使用 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 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()

也打开了 GitHub 问题：https ://github.com/spulec/moto/issues/3027

Answer 1

也许它会帮助您将所有模块放在一个类中，并在类上放置一个装饰器@mock_sns 来模拟 sns，还将装饰器@mock_sns 放在您初始化 sns 连接的函数上。例子：@mock_snsclass 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

下面的示例代码。我希望它能帮助别人。关于设置区域的建议修复不是我的问题。如果您仍然卡住，这个视频很棒。方法：创建一个模拟的 Boto3 资源（不是 Boto3 客户端）。在此新资源中设置模拟 SNS 主题 ARN。覆盖用于测试的 SNS 主题 ARN 环境变量。获取一个调用 Publish 到模拟 SNS 主题 ARN 的 Boto3 客户端。我遇到了以下错误，因为我将主题 ARN 设置为mock_topic而不是arn:aws:sns:eu-west-1:123456789012:mock_topic：botocore.errorfactory.NotFoundException：调用发布操作时发生错误（NotFound）：端点不存在“”“import mainimport boto3import pytestimport botocorefrom 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_snsdef 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 boto3import jsonimport botocoreimport osfrom conf.base_logger import logger# split the get Client and Publish for simpler testingdef 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'    )