使用Moto测试AWS邮件功能失败,请求排查解决
问题:Moto测试AWS SES发送邮件时调用真实服务并报错
背景
想用Moto测试AWS邮件发送功能,但Moto文档不完善,部分功能已变更(如原@mock_ses已移除),且对Mock技术理解较浅。
被测函数
import logging from typing import Iterable, Literal import boto3.session logger = logging.getLogger(__name__) def send_email(boto3_session: boto3.session.Session, to: Iterable[str], subject: str, body: str, body_format: Literal['Html', 'Text'] = 'Html', cc: Iterable[str] = None, from_email: str = 'test@domain.com'): # Create the email message send_args = { 'Source': from_email, 'Destination': { 'ToAddresses': to, }, 'Message': { 'Subject': {'Data': subject}, 'Body': {body_format: {'Data': body}} } } if cc: send_args['Destination']['CcAddresses'] = cc response = boto3_session.client('ses', region_name="us-east-1").send_email(**send_args) message_id = response['MessageId'] logger.debug( "Sent mail %s to %s.", message_id, to[0])
我的测试函数
import os import boto3 from moto import mock_aws from moto.core import DEFAULT_ACCOUNT_ID from moto.ses import ses_backends import pytest import pytest_check as ptc import rbn_lib.comms as comms DEFAULT_REGION = "us-east-1" @pytest.fixture def aws_credentials(): os.environ["AWS_ACCESS_KEY_ID"] = "testing" os.environ["AWS_SECRET_ACCESS_KEY"] = "testing" os.environ["AWS_SECURITY_TOKEN"] = "testing" os.environ["AWS_SESSION_TOKEN"] = "testing" os.environ["AWS_DEFAULT_REGION"] = DEFAULT_REGION @mock_aws @pytest.fixture def sess(aws_credentials): sess = boto3.Session() yield sess @mock_aws def test_send_email(sess): to = ["test@test_email.us"] subject = "Test" body = "Test" comms.send_email(sess, to, subject, body) ses_backend = ses_backends[DEFAULT_ACCOUNT_ID][DEFAULT_REGION] ptc.equal(ses_backend.sent_messages[0].subject, subject)
错误信息
错误显示实际调用了AWS真实服务,报错如下:
self = <botocore.client.SES object at 0x000002542F110310> operation_name = 'SendEmail' api_params = {'Destination': {'ToAddresses': ['test@test_email.us']}, 'Message': {'Body': {'Html': {'Data': 'Test'}}, 'Subject': {'Data': 'Test'}}, 'Source': 'test@domain.com'} def _make_api_call(self, operation_name, api_params): # ...(省略中间代码) if http.status_code >= 300: error_info = parsed_response.get("Error", {}) error_code = error_info.get("QueryErrorCode") or error_info.get( "Code" ) error_class = self.exceptions.from_code(error_code) > raise error_class(parsed_response, operation_name) E botocore.errorfactory.MessageRejected: An error occurred (MessageRejected) when calling the SendEmail operation: Email address not verified test@domain.com
解决方案
问题根源
- Mock作用域重复:测试函数和fixture都加了
@mock_aws,可能导致Mock生效异常,实际请求绕过Mock调用了真实AWS服务。 - Moto SES规则未遵守:Moto模拟了AWS SES的验证机制,发件邮箱必须先验证才能发送邮件,否则会触发和真实环境一样的报错。
- 后端获取方式过时:直接访问
ses_backends字典的方式在新版本Moto中不推荐,容易出现环境不匹配问题。
修改后的测试代码
import os import boto3 from moto import mock_aws import pytest import pytest_check as ptc import rbn_lib.comms as comms DEFAULT_REGION = "us-east-1" @pytest.fixture def aws_credentials(): os.environ["AWS_ACCESS_KEY_ID"] = "testing" os.environ["AWS_SECRET_ACCESS_KEY"] = "testing" os.environ["AWS_SECURITY_TOKEN"] = "testing" os.environ["AWS_SESSION_TOKEN"] = "testing" os.environ["AWS_DEFAULT_REGION"] = DEFAULT_REGION @mock_aws @pytest.fixture def sess(aws_credentials): # 在Mock环境内创建Session,并提前验证发件邮箱 session = boto3.Session() ses_client = session.client('ses', region_name=DEFAULT_REGION) # 验证被测函数中使用的发件邮箱 ses_client.verify_email_identity(EmailAddress='test@domain.com') yield session def test_send_email(sess): to = ["test@test_email.us"] subject = "Test" body = "Test" comms.send_email(sess, to, subject, body) # 通过客户端获取对应的后端实例,避免直接访问内部字典 ses_client = sess.client('ses', region_name=DEFAULT_REGION) ses_backend = ses_client._endpoint._url_builder._endpoint_prefix_backend ptc.equal(ses_backend.sent_messages[0].subject, subject)
关键修改点
- 移除测试函数上的
@mock_aws,只保留fixture上的装饰器,确保Mock作用域统一。 - 在fixture中提前调用
verify_email_identity验证发件邮箱,符合Moto模拟的SES规则。 - 通过客户端实例获取后端,而不是直接访问
ses_backends,适配新版本Moto的内部结构。
内容的提问来源于stack exchange,提问作者Marc
相关产品推荐
相关产品推荐

