模拟多个 boto3 服务，有些没有 moto 实现

Question

我正在尝试使用模拟对 AWS Lambda 函数中的逻辑进行单元测试。 Lambda 通过 AWS Pinpoint 发送推送通知来完成它的执行。 Lambda 还使用 AWS SSM Parameter Store。我一直在其他 Lambdas 中使用多个 boto3 对象和 moto https://github.com/spulec/moto 进行模拟，但目前在 moto 中没有 Pinpoint 实现。

我在中找到了一个解决方案，我需要对其进行修改才能使其正常工作。它回答的问题与我的确切情况无关，但答案为我指明了解决方案。所以我在这里发帖记录我对我修改的解决方案所做的更改，并询问是否有更优雅的方法来做到这一点。我看过 botocore.stub.Stubber 但看不出更好的方法，但我愿意被证明是错误的。

到目前为止我的代码：

test.py

import unittest
from unittest.mock import MagicMock, patch
import boto3
from moto import mock_ssm
import my_module


def mock_boto3_client(*args, **kwargs):
    if args[0] == 'ssm':
        # Use moto.
        mock_client = boto3.client(*args, **kwargs)
    else:
        mock_client = boto3.client(*args, **kwargs)
        if args[0] == 'pinpoint':
            # Use MagicMock.
            mock_client.create_segment = MagicMock(
                return_value={'SegmentResponse': {'Id': 'Mock SegmentID'}}
            )
            mock_client.create_campaign = MagicMock(
                return_value={'response': 'Mock Response'}
            )
    return mock_client


class TestMyModule(unittest.TestCase):
    @patch('my_module.boto3')
    @mock_ssm
    def test_my_module(self, mock_boto3):
        mock_boto3.client = mock_boto3_client
        conn = mock_boto3.client('ssm', region_name='eu-west-2')
        conn.put_parameter(
            Name='/my/test',
            Value="0123456789",
            Type='String',
            Tier='Standard'
        )
        response = my_module.handler()
        self.assertEqual(
            ('0123456789', 'Mock SegmentID', {'response': 'Mock Response'}), 
            response
        )

my_module.py

import boto3
import json


def get_parameter():
    ssm = boto3.client('ssm', region_name='eu-west-2')
    parameter = ssm.get_parameter(Name='/my/test')
    return parameter['Parameter']['Value']


def create_segment(client, message_id, push_tags, application_id):
    response = client.create_segment(
        ApplicationId=application_id,
        WriteSegmentRequest={
            'Dimensions': {
                'Attributes': {
                    'pushTags': {
                        'AttributeType': 'INCLUSIVE',
                        'Values': push_tags
                    }
                }
            },
            'Name': f'Segment {message_id}'
        }
    )
    return response['SegmentResponse']['Id']


def create_campaign(client, message_id, segment_id, application_id):
    message_payload_apns = json.dumps({
        "aps": {
            "alert": 'My Alert'
        },
        "messageId": message_id,
    })

    response = client.create_campaign(
        ApplicationId=application_id,
        WriteCampaignRequest={
            'Description': f'Test campaign - message {message_id} issued',
            'MessageConfiguration': {
                'APNSMessage': {
                    'Action': 'OPEN_APP',
                    'RawContent': message_payload_apns
                }
            },
            'Name': f'{message_id} issued',
            'Schedule': {
                'StartTime': 'IMMEDIATE'
            },
            'SegmentId': segment_id
        }
    )
    return response


def handler():
    application_id = get_parameter()
    client = boto3.client('pinpoint', region_name='eu-west-1')
    segment_id = create_segment(client, 12345, [1, 2], application_id)
    response = create_campaign(client, 12345, segment_id, application_id)
    return application_id, segment_id, response

我特别想知道如何更好、更优雅地实现 mock_boto3_client() 以更通用的方式处理。

Answer 1

对于任何新服务，使用 moto 框架都相对容易。这使您可以专注于所需的行为，而 moto 负责搭建脚手架。

在 Moto 框架中注册附加服务需要两个步骤：

确保 moto 模拟对 https://pinpoint.aws.amazon.com
创建响应 class 以响应 https://pinpoint.aws.amazon.com

模拟实际的 HTTP 请求可以通过从 moto 扩展 BaseBackend-class 来完成。请注意 urls，以及所有对此 url 的请求都将被 PinPointResponse-class.

模拟的事实

pinpoint_mock/models.py:

import re

from boto3 import Session

from moto.core import BaseBackend
from moto.sts.models import ACCOUNT_ID



class PinPointBackend(BaseBackend):

    def __init__(self, region_name):
        self.region_name = region_name

    @property
    def url_paths(self):
        return {"{0}/$": PinPointResponse.dispatch}

    @property
    def url_bases(self):
        return ["https?://pinpoint.(.+).amazonaws.com"]

    def create_app(self, name):
        # Store the app in memory, to retrieve later
        pass


pinpoint_backends = {}
for region in Session().get_available_regions("pinpoint"):
    pinpoint_backends[region] = PinPointBackend(region)
for region in Session().get_available_regions(
    "pinpoint", partition_name="aws-us-gov"
):
    pinpoint_backends[region] = PinPointBackend(region)
for region in Session().get_available_regions("pinpoint", partition_name="aws-cn"):
    pinpoint_backends[region] = PinPointBackend(region)

Response-class 需要从 moto 扩展 BaseResponse-class，并且需要复制您要模拟的方法名称。
pinpoint/responses.py

from __future__ import unicode_literals

import json

from moto.core.responses import BaseResponse
from moto.core.utils import amzn_request_id
from .models import pinpoint_backends


class PinPointResponse(BaseResponse):
    @property
    def pinpoint_backend(self):
        return pinpoint_backends[self.region]

    @amzn_request_id
    def create_app(self):
        name = self._get_param("name")
        pinpoint_backend.create_app(name)
        return 200, {}, {}

现在剩下的就是创建装饰器了：

from __future__ import unicode_literals
from .models import stepfunction_backends
from ..core.models import base_decorator

pinpoint_backend = pinpoint_backends["us-east-1"]
mock_pinpoint = base_decorator(pinpoint_backends)

@mock_pinpoint
def test():
    client = boto3.client('pinpoint')
    client.create_app(Name='testapp')

代码取自 StepFunctions 模块，它可能是最简单的模块之一，最容易适应您的需求： https://github.com/spulec/moto/tree/master/moto/stepfunctions

Answer 2

正如我在对 Bert Blommers 回答的评论中所说的那样

"I managed to register an additional service in the Moto-framework for pinpoint create_app() but failed to implement create_segment() as botocore takes "locationName": "application-id" from botocore/data/pinpoint/2016-12-01/service-2.json and then moto\core\responses.py tries to make a regex with it but creates '/v1/apps/{application-id}/segments' which has an invalid hyphen in it"

但我会 post 我的 create_app() 工作代码在这里是为了其他阅读本文的人的利益 post。

包结构很重要，因为 "pinpoint" 包需要在另一个包下。

.
├── mock_pinpoint
│   └── pinpoint
│       ├── __init__.py
│       ├── pinpoint_models.py
│       ├── pinpoint_responses.py
│       └── pinpoint_urls.py
├── my_module.py
└── test.py

mock_pinpoint/pinpoint/init.py

from __future__ import unicode_literals
from mock_pinpoint.pinpoint.pinpoint_models import pinpoint_backends
from moto.core.models import base_decorator

mock_pinpoint = base_decorator(pinpoint_backends)

mock_pinpoint/pinpoint/pinpoint_models.py

from boto3 import Session
from moto.core import BaseBackend


class PinPointBackend(BaseBackend):

    def __init__(self, region_name=None):
        self.region_name = region_name

    def create_app(self):
        # Store the app in memory, to retrieve later
        pass


pinpoint_backends = {}
for region in Session().get_available_regions("pinpoint"):
    pinpoint_backends[region] = PinPointBackend(region)

mock_pinpoint/pinpoint/pinpoint_responses.py

from __future__ import unicode_literals
import json
from moto.core.responses import BaseResponse
from mock_pinpoint.pinpoint import pinpoint_backends


class PinPointResponse(BaseResponse):
    SERVICE_NAME = "pinpoint"

    @property
    def pinpoint_backend(self):
        return pinpoint_backends[self.region]

    def create_app(self):
        body = json.loads(self.body)
        response = {
            "Arn": "arn:aws:mobiletargeting:eu-west-1:AIDACKCEVSQ6C2EXAMPLE:apps/810c7aab86d42fb2b56c8c966example",
            "Id": "810c7aab86d42fb2b56c8c966example",
            "Name": body['Name'],
            "tags": body['tags']
        }
        return 200, {}, json.dumps(response)

mock_pinpoint/pinpoint/pinpoint_urls.py

from __future__ import unicode_literals
from .pinpoint_responses import PinPointResponse

url_bases = ["https?://pinpoint.(.+).amazonaws.com"]
url_paths = {"{0}/v1/apps$": PinPointResponse.dispatch}

my_module.py

import boto3


def get_parameter():
    ssm = boto3.client('ssm', region_name='eu-west-2')
    parameter = ssm.get_parameter(Name='/my/test')
    return parameter['Parameter']['Value']


def create_app(name: str, push_tags: dict):
    client = boto3.client('pinpoint', region_name='eu-west-1')
    return client.create_app(
        CreateApplicationRequest={
            'Name': name,
            'tags': push_tags
        }
    )


def handler():
    application_id = get_parameter()
    app = create_app('my_app', {"my_tag": "tag"})
    return application_id, app

test.py

import unittest
import boto3
from moto import mock_ssm
import my_module
from mock_pinpoint.pinpoint import mock_pinpoint


class TestMyModule(unittest.TestCase):
    @mock_pinpoint
    @mock_ssm
    def test_my_module(self):
        conn = boto3.client('ssm', region_name='eu-west-2')
        conn.put_parameter(
            Name='/my/test',
            Value="0123456789",
            Type='String',
            Tier='Standard'
        )
        application_id, app = my_module.handler()
        self.assertEqual('0123456789', application_id)
        self.assertEqual(
            'arn:aws:mobiletargeting:eu-west-1:AIDACKCEVSQ6C2EXAMPLE:apps/810c7aab86d42fb2b56c8c966example',
            app['ApplicationResponse']['Arn']
        )
        self.assertEqual(
            '810c7aab86d42fb2b56c8c966example',
            app['ApplicationResponse']['Id']
        )
        self.assertEqual(
            'my_app',
            app['ApplicationResponse']['Name']
        )
        self.assertEqual(
            {"my_tag": "tag"},
            app['ApplicationResponse']['tags']
        )

话虽如此，原始问题中的解决方案有效并且更容易实现但不够优雅。

模拟多个 boto3 服务，有些没有 moto 实现

Mocking multiple boto3 services, some without moto implementation

python

amazon-web-services

mocking

boto3

moto