【问题标题】:Mocking multiple boto3 services, some without moto implementation模拟多个 boto3 服务,其中一些没有 moto 实现
【发布时间】:2020-08-01 08:49:44
【问题描述】:

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

我在https://stackoverflow.com/a/55527212/839338 中找到了一个解决方案,我需要对其进行修改才能使其正常工作。它回答的问题不是关于我的确切情况,而是答案为我指出了一个解决方案。所以我在这里发帖记录我对我修改的解决方案的更改,并询问是否有更优雅的方法来做到这一点。我查看了 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() 以更通用的方式处理。

【问题讨论】:

    标签: python amazon-web-services mocking boto3 moto


    【解决方案1】:

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

    “我设法 在 Moto 框架中注册附加服务以进行精确定位 create_app() 但未能将 create_segment() 实现为 botocore 取 "locationName": "application-id" 来自 botocore/data/pinpoint/2016-12-01/service-2.json 然后 moto\core\responses.py 尝试用它创建一个正则表达式,但创建 '/v1/apps/{application-id}/segments' 中的连字符无效 它”

    但我将在此处发布我的 create_app() 工作代码,以供阅读这篇文章的其他人受益。

    包结构很重要,因为“精确”包需要位于另一个包之下。

    .
    ├── 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']
            )
    

    话虽如此,原始问题中的解决方案有效并且更易于实施,但没有那么优雅。

    【讨论】:

      【解决方案2】:

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

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

      1. 确保 moto 模拟到 https://pinpoint.aws.amazon.com 的实际 HTTP 请求
      2. 创建一个响应类来处理https://pinpoint.aws.amazon.com 的请求

      可以通过从 moto 扩展 BaseBackend 类来模拟实际的 HTTP 请求。请注意 url,并且所有对该 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 类需要从 moto 扩展 BaseResponse 类,并且需要复制您尝试模拟的方法名称。
      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-module,它可能是更简单的模块之一,并且最容易适应您的需求: https://github.com/spulec/moto/tree/master/moto/stepfunctions

      【讨论】:

      • 我设法在 Moto 框架中为 pinpoint create_app() 注册了一项附加服务,但未能实现 create_segment(),因为 botocore 从 botocore/data/pinpoint 获取 "locationName": "application-id" /2016-12-01/service-2.json 然后 moto\core\responses.py 尝试用它创建一个正则表达式,但创建 '/v1/apps/{application-id}/segments',其中包含无效的连字符它。感谢您的帮助,但我似乎无法使用原来的解决方案。
      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2015-06-03
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多