我正在尝试使用模拟对 AWS Lambda 函数中的逻辑进行单元测试。Lambda 通过 AWS Pinpoint 发送推送通知来完成它的执行。Lambda 还使用 AWS SSM 参数存储。我一直在使用 moto https://github.com/spulec/moto在其他 Lambdas 中模拟多个 boto3 对象，但目前在 moto 中没有 Pinpoint 实现。
我在https://stackoverflow.com/a/55527212/839338中找到了一个解决方案，我需要对其进行修改才能使其正常工作。它回答的问题不是关于我的确切情况，而是答案为我指明了一个解决方案。所以我在这里发帖记录我对我修改的解决方案的更改，并询问是否有更优雅的方式来做到这一点。我看过 botocore.stub.Stubber 但看不出更好的方法，但我愿意被证明是错误的。
到目前为止我的代码：
测试.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
)
我的模块.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() 以更通用的方式处理。