Как издеваться над API SES версии 2 для boto3? - PullRequest
Новая
       26

Как издеваться над API SES версии 2 для boto3?

1 голос
/ 20 января 2020

Я пытаюсь улучшить лямбду отправителя электронной почты, чтобы она могла использовать SES AWS для отправки массовых писем, которые также имеют вложения. То, что казалось лучшим решением (в то время), было просто обновить сервис sesv2 в boto3, так как метод send_email обеспечивает необходимую функциональность.

Раньше модульные тесты просто использовали библиотеку moto для имитации служба SES, но moto в настоящее время не поддерживает sesv2. Как команда, мы изо всех сил пытаемся найти хороший способ создания рабочих юнит-тестов без использования библиотеки moto. Главным образом из-за отсутствия опытного тестировщика в команде.

Мы действительно не знаем, чтобы запустить тест, чтобы убедиться, что функция send_email работает должным образом.

sender.py 

import boto3
from email.mime.multipart import MIMEMultipart


def handler(event, context):
    msg = MIMEMultipart('mixed')
    send_raw_email(msg, 'recipient@domain.com', 'sender@domain.com')
    send_email(msg, 'recipient@domain.com', 'sender@domain.com')


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='sender@domain.com')
        result = sender.send_raw_email(MIMEMultipart('mixed'), 'recipient@domain.com', 'sender@domain.com')
        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='sender@domain.com')

        ses_client.send_email = Mock(return_value=None) 
        sender.send_email = Mock(return_value={'MessageId': 'string'})
        result = sender.send_email(MIMEMultipart('mixed'), 'recipient@domain.com', 'sender@domain.com')
        self.assertEqual(result, {'MessageId': 'string'})

1 Ответ

1 голос
/ 06 апреля 2020

У меня была такая же проблема. Вот как я решил:

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 = 'noreply@example.com'

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': ['destination@example.com'],
        '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': ['destination@example.com'],
        '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)
...