Я пытаюсь провести модульное тестирование logi c в функции AWS Lambda с помощью насмешек. Лямбда завершает свое выполнение, отправляя уведомления pu sh через 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 () для обработки более обобщенным c способом.