You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用moto的mock_ses测试AWS SES invoke_ses函数遇问题求助

AWS SES Mock测试问题排查

待测试文件 aws_ses.py

def invoke_ses(send_args, ses_client):
    try:
        response = ses_client.send_templated_email(**send_args)
    except ClientError as e:
        logging.error(e)
        return e.value.response
    else:
        return response

测试文件 test_aws_ses.py

from moto import mock_ses
from conftest import TEST_KWARGS
from emails.aws_ses import invoke_ses


def test_invoke_ses(aws_credentials, ses_client):

    # 问题1
    assert invoke_ses(TEST_KWARGS, ses_client) == ClientError

    verification_response = ses_client.verify_domain_identity(Domain="test.com")
    assert verification_response['VerificationToken'] == "QTKknzFg2J4ygwa+XvHAxUl1hyHoY0gVfZdfjIedHZ0="
    assert verification_response['ResponseMetadata']['HTTPStatusCode'] == 200
    
    # 问题2
    with pytest.raises(ClientError) as ex:
        invoke_ses(TEST_KWARGS, ses_client)
    
    assert ex.value.response["Error"]["Code"] == "TemplateDoesNotExist"

夹具文件 conftest.py

import os
import boto3
import pytest
from moto import mock_ses


DEFAULT_AWS_REGION = 'us-west-2'
SOURCE = 'superuser@test.com'
TEST_TEMPLATE_NAME = 'TEST_TEMPLATE_1'
TEST_EMAIL_RECIPIENTS = ['user1@test.com','user2@test.com','user3@test.com']
TEST_CC_LIST = ['to_cc1@test.com','to_cc2@test.com']
TEST_BCC_LIST = ['to_bcc1@test.com','to_bcc2@test.com']
TEST_SES_DESTINATION_WITH_LIMITED_EMAIL_RECIPIENTS={
            "ToAddresses": TEST_EMAIL_RECIPIENTS,
            "CcAddresses": TEST_CC_LIST,
            "BccAddresses": TEST_BCC_LIST,
        }
TEST_KWARGS = dict(
        Source=SOURCE,
        Destination=TEST_SES_DESTINATION_WITH_LIMITED_EMAIL_RECIPIENTS,
        Template=TEST_TEMPLATE_NAME,
        TemplateData='{"name": "test"}'
    )

@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"] = "us-east-1"


@mock_ses
@pytest.fixture
def ses_client():
    ses_client = boto3.client("ses", region_name=DEFAULT_AWS_REGION)
    yield ses_client

遇到的问题

问题1

执行assert invoke_ses(TEST_KWARGS, ses_client) == ClientError时抛出错误:

>       assert invoke_ses(TEST_KWARGS, ses_client) == ClientError
[CPython37-test] E       AssertionError: assert 'ClientError' == <class 'botoc....ClientError'>
[CPython37-test] E         +'ClientError'
[CPython37-test] E         -<class 'botocore.exceptions.ClientError'>
[CPython37-test] 
[CPython37-test] test/test_aws_ses.py:32: AssertionError

问题2

执行with pytest.raises(ClientError) as ex: invoke_ses(TEST_KWARGS, ses_client)时抛出错误:

>           invoke_ses(TEST_KWARGS, ses_client)
[CPython37-test] E           Failed: DID NOT RAISE <class 'botocore.exceptions.ClientError'>
[CPython37-test] 
[CPython37-test] test/test_aws_ses.py:39: Failed

注意:直接调用ses_client.send_templated_email(**TEST_KWARGS)时所有测试用例均可通过,但通过invoke_ses调用就失败。


问题原因及解决方案

核心原因

invoke_ses函数捕获了ClientError异常并返回错误响应,而非将异常抛出。这导致:

  1. 问题1:你断言返回值等于ClientError类,但实际返回的是e.value.response(包含错误信息的字典),两者完全不匹配。
  2. 问题2:pytest.raises期望函数抛出ClientError,但invoke_ses已将异常吞掉并返回响应,自然不会触发断言。

修改方案

根据业务需求选择以下两种方案:

方案1:适配当前invoke_ses的行为修改测试用例

既然函数出错时返回错误响应,就断言响应内容而非异常:

# 先补充必要导入
from botocore.exceptions import ClientError
import pytest

def test_invoke_ses(aws_credentials, ses_client):
    # 问题1修改:断言错误响应的错误码
    error_response = invoke_ses(TEST_KWARGS, ses_client)
    assert error_response["Error"]["Code"] == "TemplateDoesNotExist"

    verification_response = ses_client.verify_domain_identity(Domain="test.com")
    assert verification_response['VerificationToken'] == "QTKknzFg2J4ygwa+XvHAxUl1hyHoY0gVfZdfjIedHZ0="
    assert verification_response['ResponseMetadata']['HTTPStatusCode'] == 200
    
    # 问题2修改:同样断言返回的错误响应
    error_response = invoke_ses(TEST_KWARGS, ses_client)
    assert error_response["Error"]["Code"] == "TemplateDoesNotExist"

方案2:修改invoke_ses函数抛出异常

如果业务需要向上抛出异常,调整函数逻辑:

def invoke_ses(send_args, ses_client):
    try:
        response = ses_client.send_templated_email(**send_args)
    except ClientError as e:
        logging.error(e)
        raise  # 重新抛出捕获的异常
    else:
        return response

此时测试用例调整为:

# 补充必要导入
from botocore.exceptions import ClientError
import pytest

def test_invoke_ses(aws_credentials, ses_client):
    # 问题1修改:捕获异常并断言类型
    with pytest.raises(ClientError):
        invoke_ses(TEST_KWARGS, ses_client)

    verification_response = ses_client.verify_domain_identity(Domain="test.com")
    assert verification_response['VerificationToken'] == "QTKknzFg2J4ygwa+XvHAxUl1hyHoY0gVfZdfjIedHZ0="
    assert verification_response['ResponseMetadata']['HTTPStatusCode'] == 200
    
    # 问题2保留:捕获异常并断言错误码
    with pytest.raises(ClientError) as ex:
        invoke_ses(TEST_KWARGS, ses_client)
    
    assert ex.value.response["Error"]["Code"] == "TemplateDoesNotExist"

内容的提问来源于stack exchange,提问作者Vinit Kumar

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.09 03:55:25