如何模拟boto3的SES版本2 API？

Question

我正在尝试改进电子邮件发送器 lambda，以便它可以使用 AWS 的 SES 发送也包含附件的批量电子邮件。 （当时）最好的解决方案似乎是简单地从 boto3 的 sesv2 服务升级，因为 send_email 方法提供了所需的功能。

以前的单元测试只是使用moto库来模拟SES服务，但moto目前对sesv2的支持为零。作为一个团队，我们正在努力找出一种在不使用 moto 库的情况下创建工作单元测试的好方法。主要是因为团队中缺乏经验丰富的测试人员。

我们真的不知道如何运行测试来确保 send_email 函数按预期工作。

发件人.py

import boto3
from email.mime.multipart import MIMEMultipart


def handler(event, context):
    msg = MIMEMultipart('mixed')
    send_raw_email(msg, '[email protected]', '[email protected]')
    send_email(msg, '[email protected]', '[email protected]')


def send_raw_email(msg, recipient, sender):
    ses_client = boto3.client('ses', region_name="eu-west-1")
    response = ses_client.send_raw_email(
        Source=sender,
        Destinations=[recipient],
        RawMessage={'Data': msg.as_string()}
    )
    return response


def send_email(msg, recipient, sender):
    ses_v2_client = boto3.client('sesv2', region_name="eu-west-1")
    response = ses_v2_client.send_email(
        FromEmailAddress=sender,
        Destination={'BccAddresses': [recipient, recipient]},
        Content={'Raw': {'Data': msg.as_string()}}
    )

    return response

test_sender.py

import unittest
import boto3
from unittest.mock import Mock

from email.mime.multipart import MIMEMultipart
from moto import mock_ses

import sender


class SenderTests(unittest.TestCase):
    @mock_ses
    def test_send_raw_email(self):
        ses_client = boto3.client('ses', region_name="eu-west-1")
        ses_client.verify_email_identity(EmailAddress='[email protected]')
        result = sender.send_raw_email(MIMEMultipart('mixed'), '[email protected]', '[email protected]')
        self.assertEqual(result['ResponseMetadata']['HTTPStatusCode'], 200)

    def test_send_email(self):
        ses_client = boto3.client('sesv2', region_name="eu-west-1")
        ses_client.create_email_identity = Mock(
            return_value={'IdentityType': 'DOMAIN', 'VerifiedForSendingStatus': True})
        ses_client.create_email_identity(EmailIdentity='[email protected]')

        ses_client.send_email = Mock(return_value=None) 
        sender.send_email = Mock(return_value={'MessageId': 'string'})
        result = sender.send_email(MIMEMultipart('mixed'), '[email protected]', '[email protected]')
        self.assertEqual(result, {'MessageId': 'string'})

Answer 1

我也有同样的问题。我就是这样解决的：

发件人.py

import os
import boto3

from io import StringIO

from botocore.exceptions import ClientError

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication


class SESHandler(object):
    def __init__(
            self,
            aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', ''),
            aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', ''),
            aws_region_name=os.environ.get('AWS_DEFAULT_REGION', 'eu-west-1'),
            client=None
    ):
        self.aws_access_key_id = aws_access_key_id
        self.aws_secret_access_key = aws_secret_access_key
        self.aws_region_name = aws_region_name
        self.client = client or self.setup_client()

    def setup_client(self):
        return boto3.client('ses', region_name=self.aws_region_name)

    def send_email(
            self, source, destination, subject, message, attachments=None
    ):
        charset = "utf-8"

        msg = MIMEMultipart('mixed')

        msg['Subject'] = subject

        msg_body = MIMEMultipart('alternative')
        msg_body.attach(MIMEText(message.encode(charset), 'plain', charset))

        msg.attach(msg_body)

        if attachments:
            for request in attachments:
                msg.attach(self.create_attachment(**request))

        try:
            response = self.client.send_raw_email(
                Source=source,
                Destinations=destination,
                RawMessage={
                    'Data': msg.as_string(),
                }
            )
        except ClientError as e:
            return e.response
        else:
            return response

    @staticmethod
    def create_attachment(file_name, file_content):
        str_io = StringIO(file_content)

        attachment = MIMEApplication(str_io.getvalue())
        attachment.add_header(
            'Content-Disposition', 'attachment', filename=file_name
        )

        return attachment

test_sender.py

import boto3
from ne_pki_service.lib.aws_tools.ses_handler import SESHandler
from moto import (
    mock_ses
)

# RequestId used in <from moto.ses.responses import END_RAW_EMAIL_RESPONSE>
_request_id = 'e0abcdfa-c866-11e0-b6d0-273d09173b49'

_aws_region = 'eu-west-1'
_aws_access_key = 'access_key'
_aws_secret_key = 'secret_key'

_source_email = '[email protected]'

boto3.setup_default_session(region_name=_aws_region)


def test_setup_client():
    my_handler = SESHandler(aws_access_key_id=_aws_access_key,
                            aws_secret_access_key=_aws_secret_key,
                            aws_region_name=_aws_region)
    assert my_handler.aws_access_key_id == _aws_access_key
    assert my_handler.aws_secret_access_key == _aws_secret_key
    assert my_handler.aws_region_name == _aws_region


@mock_ses
def test_send_email_success():
    client = boto3.client('ses', _aws_region)

    client.verify_email_identity(EmailAddress=_source_email)

    handler = SESHandler(client=client)
    data = {
        'source': _source_email,
        'destination': ['[email protected]'],
        'subject': 'Test subject',
        'message': 'Test message',
        'attachments': [
            {
                'file_name': 'token.dat',
                'file_content': 'TOKEN-TEST'
            }
        ]
    }
    response = handler.send_email(**data)

    assert response['ResponseMetadata']['HTTPStatusCode'] == 200
    assert response['ResponseMetadata']['RequestId'] == _request_id


@mock_ses
def test_send_email_failure():
    client = boto3.client('ses', _aws_region)

    handler = SESHandler(client=client)
    data = {
        'source': _source_email,
        'destination': ['[email protected]'],
        'subject': 'Test subject',
        'message': 'Test message',
    }
    response = handler.send_email(**data)

    assert response['ResponseMetadata']['HTTPStatusCode'] == 400
    assert response['Error']['Message'] == \
        'Did not have authority to send from email {}'.format(_source_email)

Answer 2

我也遇到了这个问题，在这个答案的日期，这就是我使用 moto 嘲笑 sesV2 的方式。
这是我要模拟的代码。

def send_email_to_client(to_email_address: str, email_content: dict):
    ses_client = boto3.client('sesv2', endpoint_url=SES_ENDPOINT_URL, region_name=AWS_REGION_NAME)
    ses_client.send_email(
        FromEmailAddress=DEFAULT_EMAIL_SENDER,
        Destination={
            'ToAddresses': [
                to_email_address
            ]
        },
        Content={
            'Simple': {
                'Subject': {
                    'Data': email_content["subject"]
                },
                'Body': {
                    'Text': {
                        'Data': email_content["body"]
                    }
                }
            }
        }
    )

这是我如何嘲笑对 sesV2 的调用：

from moto.core import DEFAULT_ACCOUNT_ID
from moto.ses import ses_backends
from config import AWS_REGION_NAME


@pytest.fixture
def ses_v1():
    with moto.mock_ses():
        yield boto3.client('ses', region_name=AWS_REGION_NAME)

@pytest.fixture
def ses_v2(ses_v1):
    ses_v1.verify_domain_identity(Domain='email.com')
    with moto.mock_sesv2():
        yield

def test_send_email_to_client():
    send_email_to_client(
        [email protected],
        email_content={'subject': 'Email subject', 'body': 'Email body'}
    )
    assert ses_backends[DEFAULT_ACCOUNT_ID][AWS_REGION_NAME].sent_messages[0].subject == 'Email subject'
    assert ses_backends[DEFAULT_ACCOUNT_ID][AWS_REGION_NAME].sent_messages[0].body == 'Email body'

技巧是使用 moto sesV2 模拟呼叫，但使用 sesV1 进行其他操作（验证电子邮件/域，检查后端以查看已发送的电子邮件）