[Python] Unit testing Amazon SES in Django: emails not being sent

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.

aram 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
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.com/ses/latest/dg/send-email-smtp-client-command-line.html

Any advice would be appreciated.

Continue reading...