Creating unit tests for Amazon Simple Email Service (SES) for a Django application using package django-ses
test_mail.py
from django.core import mail
...
def test_send_direct_email(send_ct):
from_email = settings.SERVER_EMAIL
to_email = [nt[2] for nt in settings.NOTIFICATIONS_TESTERS]
starttime = datetime.now()
connection = mail.get_connection()
pre_data = get_ses_emails_data()
_mail_signal_assertion_handler.call_count = 0
signals.message_sent.connect(_mail_signal_assertion_handler)
emails = []
for i in range(send_ct):
emails.append(
mail.EmailMessage(
SUBJECT_EMAIL,
BODY_EMAIL.format(send_ct=i, server=settings.EMAIL_BACKEND),
from_email,
to_email,
# connection=connection,
)
)
connection.send_messages(emails)
post_data = get_ses_emails_data()
assert int(post_data["24hour_sent"]) == int(pre_data["24hour_sent"]) + send_ct
assert check_aws_ses_sent(assertions={"Sent": send_ct, "StartTime": starttime})
assert _mail_signal_assertion_handler.call_count == send_ct
settings.py
AWS_DEFAULT_REGION = "ca-central-1"
try:
# IAM programmatic user
AWS_ACCESS_KEY_ID = env("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = env("AWS_SECRET_ACCESS_KEY")
except KeyError:
raise ImproperlyConfigured("Missing AWS_ACCESS_KEY_ID or AWS_SECRET_ACCESS_KEY")
# =========== EMAIL ==============
EMAIL_BACKEND = "django_ses.SESBackend"
DEFAULT_FROM_EMAIL = env("DEFAULT_FROM_EMAIL") # verified aws ses identity
SERVER_EMAIL = DEFAULT_FROM_EMAIL
but the emails are never sent (AssertionFrror: False (0 == 1). The service is working as expected when running live on the server.
The assertions I am using are a connection to the message_sent signal (new in 4.4.0)
from django_ses import signals
def _mail_signal_assertion_handler(sender, message, **kwargs):
_mail_signal_assertion_handler.call_count += 1
assert message.subject == SUBJECT_EMAIL
assert message.body == BODY_EMAIL.format(
send_ct=_mail_signal_assertion_handler.call_count, server=settings.EMAIL_BACKEND
)
signals.message_sent.connect(_mail_signal_assertion_handler)
and checking the SES data through a boto3 client session:
from django_ses.views import emails_parse, stats_to_list, sum_stats
def get_ses_emails_data(ses_conn=None) -> dict[str, Any]:
try:
if not ses_conn:
ses_conn = boto3.client(
"ses",
aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
region_name=settings.AWS_DEFAULT_REGION,
)
quota_dict = ses_conn.get_send_quota()
verified_emails_dict = ses_conn.list_verified_email_addresses()
stats = ses_conn.get_send_statistics()
verified_emails = emails_parse(verified_emails_dict)
ordered_data = stats_to_list(stats)
summary = sum_stats(ordered_data)
return {
"datapoints": ordered_data,
"24hour_quota": quota_dict["Max24HourSend"],
"24hour_sent": quota_dict["SentLast24Hours"],
"24hour_remaining": quota_dict["Max24HourSend"] - quota_dict["SentLast24Hours"],
"persecond_rate": quota_dict["MaxSendRate"],
"verified_emails": verified_emails,
"summary": summary,
}
except Exception as e:
import traceback
traceback.print_exc()
print(e)
raise
def check_aws_ses_sent(assertions: dict[str, Any]) -> bool:
"""
Check if the required number of emails were sent within the given time frame.
:param assertions: Dictionary with "Sent" (number of emails to check) and "StartTime" (datetime).
:return: True if all assertions pass, otherwise raises an AssertionError.
"""
email_data = get_ses_emails_data()
emails_to_check = assertions["Sent"]
per_second_rate = int(email_data["persecond_rate"])
datapoints = list(reversed(email_data["datapoints"]))
# Calculate the number of datapoints to check and the remainder
required_datapoints = int(emails_to_check / per_second_rate)
remainder = int(emails_to_check % per_second_rate)
if required_datapoints == 0:
# Handle the case where the number of emails to check is less than the per-second rate
remainder = 0
required_datapoints = 1
per_second_rate = emails_to_check
for i in range(required_datapoints):
if i >= len(datapoints):
raise AssertionError("Not enough datapoints to validate email sending assertions.")
datapoint = datapoints[i]
dp_timestamp = normalize_aws_timestamp(datapoint["Timestamp"])
timestamp = normalize_djmail_timestamp(assertions["StartTime"])
# Check timestamp
assert (
dp_timestamp >= timestamp
), f"Datapoint at index {i} has timestamp {dp_timestamp} before the start time {timestamp}."
# Check delivery attempts
expected_attempts = remainder if i == 0 and remainder > 0 else per_second_rate
assert int(datapoint["DeliveryAttempts"]) == int(
expected_attempts
), f"Datapoint at index {i} has {datapoint['DeliveryAttempts']} attempts; expected {expected_attempts}."
# Remainder handled only for the first datapoint
remainder = 0
return True
I've tried all the recommended ways to open a connection, create an EmailMessage object, and send it.
I haven't tried instancing the SESBackend directly itself rather using mail.get_connection() but I don't think I should have to.
I have a good connection to the AWS mail server, as per .html
Any advice would be appreciated.
Creating unit tests for Amazon Simple Email Service (SES) for a Django application using package django-ses
test_mail.py
from django.core import mail
...
def test_send_direct_email(send_ct):
from_email = settings.SERVER_EMAIL
to_email = [nt[2] for nt in settings.NOTIFICATIONS_TESTERS]
starttime = datetime.now()
connection = mail.get_connection()
pre_data = get_ses_emails_data()
_mail_signal_assertion_handler.call_count = 0
signals.message_sent.connect(_mail_signal_assertion_handler)
emails = []
for i in range(send_ct):
emails.append(
mail.EmailMessage(
SUBJECT_EMAIL,
BODY_EMAIL.format(send_ct=i, server=settings.EMAIL_BACKEND),
from_email,
to_email,
# connection=connection,
)
)
connection.send_messages(emails)
post_data = get_ses_emails_data()
assert int(post_data["24hour_sent"]) == int(pre_data["24hour_sent"]) + send_ct
assert check_aws_ses_sent(assertions={"Sent": send_ct, "StartTime": starttime})
assert _mail_signal_assertion_handler.call_count == send_ct
settings.py
AWS_DEFAULT_REGION = "ca-central-1"
try:
# IAM programmatic user
AWS_ACCESS_KEY_ID = env("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = env("AWS_SECRET_ACCESS_KEY")
except KeyError:
raise ImproperlyConfigured("Missing AWS_ACCESS_KEY_ID or AWS_SECRET_ACCESS_KEY")
# =========== EMAIL ==============
EMAIL_BACKEND = "django_ses.SESBackend"
DEFAULT_FROM_EMAIL = env("DEFAULT_FROM_EMAIL") # verified aws ses identity
SERVER_EMAIL = DEFAULT_FROM_EMAIL
but the emails are never sent (AssertionFrror: False (0 == 1). The service is working as expected when running live on the server.
The assertions I am using are a connection to the message_sent signal (new in 4.4.0)
from django_ses import signals
def _mail_signal_assertion_handler(sender, message, **kwargs):
_mail_signal_assertion_handler.call_count += 1
assert message.subject == SUBJECT_EMAIL
assert message.body == BODY_EMAIL.format(
send_ct=_mail_signal_assertion_handler.call_count, server=settings.EMAIL_BACKEND
)
signals.message_sent.connect(_mail_signal_assertion_handler)
and checking the SES data through a boto3 client session:
from django_ses.views import emails_parse, stats_to_list, sum_stats
def get_ses_emails_data(ses_conn=None) -> dict[str, Any]:
try:
if not ses_conn:
ses_conn = boto3.client(
"ses",
aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
region_name=settings.AWS_DEFAULT_REGION,
)
quota_dict = ses_conn.get_send_quota()
verified_emails_dict = ses_conn.list_verified_email_addresses()
stats = ses_conn.get_send_statistics()
verified_emails = emails_parse(verified_emails_dict)
ordered_data = stats_to_list(stats)
summary = sum_stats(ordered_data)
return {
"datapoints": ordered_data,
"24hour_quota": quota_dict["Max24HourSend"],
"24hour_sent": quota_dict["SentLast24Hours"],
"24hour_remaining": quota_dict["Max24HourSend"] - quota_dict["SentLast24Hours"],
"persecond_rate": quota_dict["MaxSendRate"],
"verified_emails": verified_emails,
"summary": summary,
}
except Exception as e:
import traceback
traceback.print_exc()
print(e)
raise
def check_aws_ses_sent(assertions: dict[str, Any]) -> bool:
"""
Check if the required number of emails were sent within the given time frame.
:param assertions: Dictionary with "Sent" (number of emails to check) and "StartTime" (datetime).
:return: True if all assertions pass, otherwise raises an AssertionError.
"""
email_data = get_ses_emails_data()
emails_to_check = assertions["Sent"]
per_second_rate = int(email_data["persecond_rate"])
datapoints = list(reversed(email_data["datapoints"]))
# Calculate the number of datapoints to check and the remainder
required_datapoints = int(emails_to_check / per_second_rate)
remainder = int(emails_to_check % per_second_rate)
if required_datapoints == 0:
# Handle the case where the number of emails to check is less than the per-second rate
remainder = 0
required_datapoints = 1
per_second_rate = emails_to_check
for i in range(required_datapoints):
if i >= len(datapoints):
raise AssertionError("Not enough datapoints to validate email sending assertions.")
datapoint = datapoints[i]
dp_timestamp = normalize_aws_timestamp(datapoint["Timestamp"])
timestamp = normalize_djmail_timestamp(assertions["StartTime"])
# Check timestamp
assert (
dp_timestamp >= timestamp
), f"Datapoint at index {i} has timestamp {dp_timestamp} before the start time {timestamp}."
# Check delivery attempts
expected_attempts = remainder if i == 0 and remainder > 0 else per_second_rate
assert int(datapoint["DeliveryAttempts"]) == int(
expected_attempts
), f"Datapoint at index {i} has {datapoint['DeliveryAttempts']} attempts; expected {expected_attempts}."
# Remainder handled only for the first datapoint
remainder = 0
return True
I've tried all the recommended ways to open a connection, create an EmailMessage object, and send it.
I haven't tried instancing the SESBackend directly itself rather using mail.get_connection() but I don't think I should have to.
I have a good connection to the AWS mail server, as per https://docs.aws.amazon/ses/latest/dg/send-email-smtp-client-command-line.html
Any advice would be appreciated.
Per https://docs.djangoproject/en/5.1/topics/testing/tools/#email-services
If any of your Django views send email using Django’s email functionality, you probably don’t want to send email each time you run a test using that view. For this reason, Django’s test runner automatically redirects all Django-sent email to a dummy outbox. This lets you test every aspect of sending email – from the number of messages sent to the contents of each message – without actually sending the messages.
The test runner accomplishes this by transparently replacing the normal email backend with a testing backend. (Don’t worry – this has no effect on any other email senders outside of Django, such as your machine’s mail server, if you’re running one.)
In other words, I think this is expected behavior; you should look at django.core.mail.outbox and see your emails.
I haven't tested it, but it appears you can do something like:
@override_settings(EMAIL_BACKEND="django_ses.SESBackend")
def test_send_direct_email(send_ct):
...
in order to override Django's default test behavior and actually use the SESBackend in tests (I'm not sure, you might need to also override some of the other settings but I think that's the main one for this use case)
Are you sure you are using the correct SES credentials? They are not the same as the typical AWS access key and secret - the SES is an additional set of credentials.
For example, I have an app that needs access to S3 to process files, and the same app also sends emails via SES.
I have to load the ACCESS KEY and SECRET KEY, in order to get the IAM permissions to read and write from S3, but I also have to load a different set of credentials (specific to SES), to use as the username and password to send emails via SES using SMTP.
I would goto the SES console, get the SES specific credentials, hard code them into your script above and see if that solves the problem (and if it does, then figure out how to load those from a config file, i.e. don't permanently hardcode them)