How to mock boto3's SES Version 2 API?

Question

I'm attempting to improve an email sender lambda so that it can use AWS's SES to send bulk emails that also have an attachment. What seemed like the best solution (at the time) was to simply upgrade from to boto3's sesv2 service, as the send_email method provides the functionality that is needed.

Previously the unit tests were simply using the moto library to mock the SES service, but moto currently has zero support for sesv2. As a team, we're struggling to figure out a good way to create working unit tests without using the moto library. Mostly due to lack of an experienced tester in the team.

We really don't know to run a test to ensure that the send_email function is working as intended.

sender.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

I had the same issue. That's how I solved:

sender.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)