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.

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

Any advice would be appreciated.

Вернуться на верх