mirror of
https://github.com/django/django.git
synced 2024-12-23 01:25:58 +00:00
2511 lines
95 KiB
Python
2511 lines
95 KiB
Python
import mimetypes
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import sys
|
||
import tempfile
|
||
from email import charset, message_from_binary_file
|
||
from email import message_from_bytes as _message_from_bytes
|
||
from email import policy
|
||
from email.message import EmailMessage as PyEmailMessage
|
||
from email.message import Message as PyMessage
|
||
from email.mime.image import MIMEImage
|
||
from email.mime.text import MIMEText
|
||
from io import StringIO
|
||
from pathlib import Path
|
||
from smtplib import SMTP, SMTPException
|
||
from ssl import SSLError
|
||
from textwrap import dedent
|
||
from unittest import mock, skipUnless
|
||
|
||
from django.core import mail
|
||
from django.core.mail import (
|
||
DNS_NAME,
|
||
BadHeaderError,
|
||
EmailAlternative,
|
||
EmailAttachment,
|
||
EmailMessage,
|
||
EmailMultiAlternatives,
|
||
mail_admins,
|
||
mail_managers,
|
||
send_mail,
|
||
send_mass_mail,
|
||
)
|
||
from django.core.mail.backends import console, dummy, filebased, locmem, smtp
|
||
from django.core.mail.message import sanitize_address
|
||
from django.test import SimpleTestCase, override_settings
|
||
from django.test.utils import requires_tz_support
|
||
from django.utils.translation import gettext_lazy
|
||
|
||
try:
|
||
from aiosmtpd.controller import Controller
|
||
|
||
HAS_AIOSMTPD = True
|
||
except ImportError:
|
||
HAS_AIOSMTPD = False
|
||
|
||
|
||
def message_from_bytes(s):
|
||
"""
|
||
email.message_from_bytes() using modern email.policy.default.
|
||
Returns a modern email.message.EmailMessage.
|
||
"""
|
||
return _message_from_bytes(s, policy=policy.default)
|
||
|
||
|
||
class MailTestsMixin:
|
||
def assertMessageHasHeaders(self, message, headers):
|
||
"""
|
||
Asserts that the `message` has all `headers`.
|
||
|
||
message: can be an instance of an email.Message subclass or bytes
|
||
with the contents of an email message.
|
||
headers: should be a set of (header-name, header-value) tuples.
|
||
"""
|
||
if isinstance(message, bytes):
|
||
message = message_from_bytes(message)
|
||
msg_headers = set(message.items())
|
||
if not headers.issubset(msg_headers):
|
||
missing = "\n".join(f" {h}: {v}" for h, v in headers - msg_headers)
|
||
actual = "\n".join(f" {h}: {v}" for h, v in msg_headers)
|
||
raise self.failureException(
|
||
f"Expected headers not found in message.\n"
|
||
f"Missing headers:\n{missing}\n"
|
||
f"Actual headers:\n{actual}"
|
||
)
|
||
|
||
# In assertStartsWith()/assertEndsWith() failure messages, when truncating
|
||
# a long first ("haystack") string, include this many characters beyond the
|
||
# length of the second ("needle") string.
|
||
START_END_EXTRA_CONTEXT = 15
|
||
|
||
def assertStartsWith(self, first, second):
|
||
if not first.startswith(second):
|
||
# Use assertEqual() for failure message with diffs. If first value
|
||
# is much longer than second, truncate end and add an ellipsis.
|
||
self.longMessage = True
|
||
max_len = len(second) + self.START_END_EXTRA_CONTEXT
|
||
start_of_first = (
|
||
first
|
||
if len(first) <= max_len
|
||
else first[:max_len] + ("…" if isinstance(first, str) else b"...")
|
||
)
|
||
self.assertEqual(
|
||
start_of_first,
|
||
second,
|
||
"First string doesn't start with the second.",
|
||
)
|
||
|
||
def assertEndsWith(self, first, second):
|
||
if not first.endswith(second):
|
||
# Use assertEqual() for failure message with diffs. If first value
|
||
# is much longer than second, truncate start and prepend an ellipsis.
|
||
self.longMessage = True
|
||
max_len = len(second) + self.START_END_EXTRA_CONTEXT
|
||
end_of_first = (
|
||
first
|
||
if len(first) <= max_len
|
||
else ("…" if isinstance(first, str) else b"...") + first[-max_len:]
|
||
)
|
||
self.assertEqual(
|
||
end_of_first,
|
||
second,
|
||
"First string doesn't end with the second.",
|
||
)
|
||
|
||
def get_raw_attachments(self, django_message):
|
||
"""
|
||
Return a list of the raw attachment parts in the MIME message generated
|
||
by serializing django_message and reparsing the result.
|
||
|
||
This returns only "top-level" attachments. It will not descend into
|
||
message/* attached emails to find nested attachments.
|
||
"""
|
||
msg_bytes = django_message.message().as_bytes()
|
||
message = message_from_bytes(msg_bytes)
|
||
return list(message.iter_attachments())
|
||
|
||
def get_decoded_attachments(self, django_message):
|
||
"""
|
||
Return a list of decoded attachments resulting from serializing
|
||
django_message and reparsing the result.
|
||
|
||
Each attachment is returned as an EmailAttachment named tuple with
|
||
fields filename, content, and mimetype. The content will be decoded
|
||
to str for mimetype text/*; retained as bytes for other mimetypes.
|
||
"""
|
||
return [
|
||
EmailAttachment(
|
||
attachment.get_filename(),
|
||
attachment.get_content(),
|
||
attachment.get_content_type(),
|
||
)
|
||
for attachment in self.get_raw_attachments(django_message)
|
||
]
|
||
|
||
def get_message_structure(self, message, level=0):
|
||
"""
|
||
Return a multiline indented string representation
|
||
of the message's MIME content-type structure, e.g.:
|
||
|
||
multipart/mixed
|
||
multipart/alternative
|
||
text/plain
|
||
text/html
|
||
image/jpg
|
||
text/calendar
|
||
"""
|
||
# Adapted from email.iterators._structure().
|
||
indent = " " * (level * 4)
|
||
structure = [f"{indent}{message.get_content_type()}\n"]
|
||
if message.is_multipart():
|
||
for subpart in message.get_payload():
|
||
structure.append(self.get_message_structure(subpart, level + 1))
|
||
return "".join(structure)
|
||
|
||
|
||
class MailTests(MailTestsMixin, SimpleTestCase):
|
||
"""
|
||
Non-backend specific tests.
|
||
"""
|
||
|
||
def test_ascii(self):
|
||
email = EmailMessage(
|
||
"Subject", "Content", "from@example.com", ["to@example.com"]
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message["Subject"], "Subject")
|
||
self.assertEqual(message.get_payload(), "Content")
|
||
self.assertEqual(message["From"], "from@example.com")
|
||
self.assertEqual(message["To"], "to@example.com")
|
||
|
||
@mock.patch("django.core.mail.message.MIMEText.set_payload")
|
||
def test_nonascii_as_string_with_ascii_charset(self, mock_set_payload):
|
||
"""Line length check should encode the payload supporting `surrogateescape`.
|
||
|
||
Following https://github.com/python/cpython/issues/76511, newer
|
||
versions of Python (3.11.9, 3.12.3 and 3.13) ensure that a message's
|
||
payload is encoded with the provided charset and `surrogateescape` is
|
||
used as the error handling strategy.
|
||
|
||
This test is heavily based on the test from the fix for the bug above.
|
||
Line length checks in SafeMIMEText's set_payload should also use the
|
||
same error handling strategy to avoid errors such as:
|
||
|
||
UnicodeEncodeError: 'utf-8' codec can't encode <...>: surrogates not allowed
|
||
|
||
"""
|
||
|
||
def simplified_set_payload(instance, payload, charset):
|
||
instance._payload = payload
|
||
|
||
mock_set_payload.side_effect = simplified_set_payload
|
||
|
||
text = (
|
||
"Text heavily based in Python's text for non-ascii messages: Föö bär"
|
||
).encode("iso-8859-1")
|
||
body = text.decode("ascii", errors="surrogateescape")
|
||
email = EmailMessage("Subject", body, "from@example.com", ["to@example.com"])
|
||
message = email.message()
|
||
mock_set_payload.assert_called_once()
|
||
self.assertEqual(message.get_payload(decode=True), text)
|
||
|
||
def test_multiple_recipients(self):
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com", "other@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message["Subject"], "Subject")
|
||
self.assertEqual(message.get_payload(), "Content")
|
||
self.assertEqual(message["From"], "from@example.com")
|
||
self.assertEqual(message["To"], "to@example.com, other@example.com")
|
||
|
||
def test_header_omitted_for_no_to_recipients(self):
|
||
message = EmailMessage(
|
||
"Subject", "Content", "from@example.com", cc=["cc@example.com"]
|
||
).message()
|
||
self.assertNotIn("To", message)
|
||
|
||
def test_recipients_with_empty_strings(self):
|
||
"""
|
||
Empty strings in various recipient arguments are always stripped
|
||
off the final recipient list.
|
||
"""
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com", ""],
|
||
cc=["cc@example.com", ""],
|
||
bcc=["", "bcc@example.com"],
|
||
reply_to=["", None],
|
||
)
|
||
self.assertEqual(
|
||
email.recipients(), ["to@example.com", "cc@example.com", "bcc@example.com"]
|
||
)
|
||
|
||
def test_cc(self):
|
||
"""Regression test for #7722"""
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com"],
|
||
cc=["cc@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message["Cc"], "cc@example.com")
|
||
self.assertEqual(email.recipients(), ["to@example.com", "cc@example.com"])
|
||
|
||
# Test multiple CC with multiple To
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com", "other@example.com"],
|
||
cc=["cc@example.com", "cc.other@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message["Cc"], "cc@example.com, cc.other@example.com")
|
||
self.assertEqual(
|
||
email.recipients(),
|
||
[
|
||
"to@example.com",
|
||
"other@example.com",
|
||
"cc@example.com",
|
||
"cc.other@example.com",
|
||
],
|
||
)
|
||
|
||
# Testing with Bcc
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com", "other@example.com"],
|
||
cc=["cc@example.com", "cc.other@example.com"],
|
||
bcc=["bcc@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message["Cc"], "cc@example.com, cc.other@example.com")
|
||
self.assertEqual(
|
||
email.recipients(),
|
||
[
|
||
"to@example.com",
|
||
"other@example.com",
|
||
"cc@example.com",
|
||
"cc.other@example.com",
|
||
"bcc@example.com",
|
||
],
|
||
)
|
||
|
||
def test_cc_headers(self):
|
||
message = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"bounce@example.com",
|
||
["to@example.com"],
|
||
cc=["foo@example.com"],
|
||
headers={"Cc": "override@example.com"},
|
||
).message()
|
||
self.assertEqual(message.get_all("Cc"), ["override@example.com"])
|
||
|
||
def test_cc_in_headers_only(self):
|
||
message = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"bounce@example.com",
|
||
["to@example.com"],
|
||
headers={"Cc": "foo@example.com"},
|
||
).message()
|
||
self.assertEqual(message.get_all("Cc"), ["foo@example.com"])
|
||
|
||
def test_bcc_not_in_headers(self):
|
||
"""
|
||
A bcc address should be in the recipients,
|
||
but not in the (visible) message headers.
|
||
"""
|
||
email = EmailMessage(
|
||
to=["to@example.com"],
|
||
bcc=["bcc@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertNotIn("Bcc", message)
|
||
self.assertNotIn("bcc@example.com", message.as_string())
|
||
self.assertEqual(email.recipients(), ["to@example.com", "bcc@example.com"])
|
||
|
||
def test_reply_to(self):
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com"],
|
||
reply_to=["reply_to@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message["Reply-To"], "reply_to@example.com")
|
||
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com"],
|
||
reply_to=["reply_to1@example.com", "reply_to2@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(
|
||
message["Reply-To"], "reply_to1@example.com, reply_to2@example.com"
|
||
)
|
||
|
||
def test_recipients_as_tuple(self):
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
("to@example.com", "other@example.com"),
|
||
cc=("cc@example.com", "cc.other@example.com"),
|
||
bcc=("bcc@example.com",),
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message["Cc"], "cc@example.com, cc.other@example.com")
|
||
self.assertEqual(
|
||
email.recipients(),
|
||
[
|
||
"to@example.com",
|
||
"other@example.com",
|
||
"cc@example.com",
|
||
"cc.other@example.com",
|
||
"bcc@example.com",
|
||
],
|
||
)
|
||
|
||
def test_recipients_as_string(self):
|
||
with self.assertRaisesMessage(
|
||
TypeError, '"to" argument must be a list or tuple'
|
||
):
|
||
EmailMessage(to="foo@example.com")
|
||
with self.assertRaisesMessage(
|
||
TypeError, '"cc" argument must be a list or tuple'
|
||
):
|
||
EmailMessage(cc="foo@example.com")
|
||
with self.assertRaisesMessage(
|
||
TypeError, '"bcc" argument must be a list or tuple'
|
||
):
|
||
EmailMessage(bcc="foo@example.com")
|
||
with self.assertRaisesMessage(
|
||
TypeError, '"reply_to" argument must be a list or tuple'
|
||
):
|
||
EmailMessage(reply_to="reply_to@example.com")
|
||
|
||
def test_header_injection(self):
|
||
msg = "Header values can't contain newlines "
|
||
cases = [
|
||
{"subject": "Subject\nInjection Test"},
|
||
{"subject": gettext_lazy("Lazy Subject\nInjection Test")},
|
||
{"to": ["Name\nInjection test <to@example.com>"]},
|
||
]
|
||
for kwargs in cases:
|
||
with self.subTest(case=kwargs):
|
||
email = EmailMessage(**kwargs)
|
||
with self.assertRaisesMessage(BadHeaderError, msg):
|
||
email.message()
|
||
|
||
def test_folding_white_space(self):
|
||
"""
|
||
Test for correct use of "folding white space" in long headers (#7747)
|
||
"""
|
||
email = EmailMessage(
|
||
"Long subject lines that get wrapped should contain a space continuation "
|
||
"character to get expected behavior in Outlook and Thunderbird",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(
|
||
message["Subject"].encode(),
|
||
b"Long subject lines that get wrapped should contain a space continuation\n"
|
||
b" character to get expected behavior in Outlook and Thunderbird",
|
||
)
|
||
|
||
def test_message_header_overrides(self):
|
||
"""
|
||
Specifying dates or message-ids in the extra headers overrides the
|
||
default values (#9233)
|
||
"""
|
||
headers = {"date": "Fri, 09 Nov 2001 01:08:47 -0000", "Message-ID": "foo"}
|
||
email = EmailMessage(headers=headers)
|
||
|
||
self.assertMessageHasHeaders(
|
||
email.message(),
|
||
{
|
||
("Message-ID", "foo"),
|
||
("date", "Fri, 09 Nov 2001 01:08:47 -0000"),
|
||
},
|
||
)
|
||
|
||
def test_from_header(self):
|
||
"""
|
||
Make sure we can manually set the From header (#9214)
|
||
"""
|
||
email = EmailMessage(
|
||
from_email="bounce@example.com",
|
||
headers={"From": "from@example.com"},
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message.get_all("From"), ["from@example.com"])
|
||
|
||
def test_to_header(self):
|
||
"""
|
||
Make sure we can manually set the To header (#17444)
|
||
"""
|
||
email = EmailMessage(
|
||
to=["list-subscriber@example.com", "list-subscriber2@example.com"],
|
||
headers={"To": "mailing-list@example.com"},
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message.get_all("To"), ["mailing-list@example.com"])
|
||
self.assertEqual(
|
||
email.to, ["list-subscriber@example.com", "list-subscriber2@example.com"]
|
||
)
|
||
|
||
# If we don't set the To header manually, it should default to the `to`
|
||
# argument to the constructor.
|
||
email = EmailMessage(
|
||
to=["list-subscriber@example.com", "list-subscriber2@example.com"],
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(
|
||
message.get_all("To"),
|
||
["list-subscriber@example.com, list-subscriber2@example.com"],
|
||
)
|
||
self.assertEqual(
|
||
email.to, ["list-subscriber@example.com", "list-subscriber2@example.com"]
|
||
)
|
||
|
||
def test_to_in_headers_only(self):
|
||
message = EmailMessage(
|
||
headers={"To": "to@example.com"},
|
||
).message()
|
||
self.assertEqual(message.get_all("To"), ["to@example.com"])
|
||
|
||
def test_reply_to_header(self):
|
||
"""
|
||
Specifying 'Reply-To' in headers should override reply_to.
|
||
"""
|
||
email = EmailMessage(
|
||
reply_to=["foo@example.com"],
|
||
headers={"Reply-To": "override@example.com"},
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message.get_all("Reply-To"), ["override@example.com"])
|
||
|
||
def test_reply_to_in_headers_only(self):
|
||
message = EmailMessage(
|
||
headers={"Reply-To": "reply_to@example.com"},
|
||
).message()
|
||
self.assertEqual(message.get_all("Reply-To"), ["reply_to@example.com"])
|
||
|
||
def test_multiple_message_call(self):
|
||
"""
|
||
Regression for #13259 - Make sure that headers are not changed when
|
||
calling EmailMessage.message()
|
||
"""
|
||
email = EmailMessage(
|
||
from_email="bounce@example.com",
|
||
headers={"From": "from@example.com"},
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message.get_all("From"), ["from@example.com"])
|
||
message = email.message()
|
||
self.assertEqual(message.get_all("From"), ["from@example.com"])
|
||
|
||
def test_unicode_address_header(self):
|
||
"""
|
||
Regression for #11144 - When a to/from/cc header contains Unicode,
|
||
make sure the email addresses are parsed correctly (especially with
|
||
regards to commas)
|
||
"""
|
||
email = EmailMessage(
|
||
to=['"Firstname Sürname" <to@example.com>', "other@example.com"],
|
||
)
|
||
self.assertEqual(
|
||
email.message()["To"],
|
||
"=?utf-8?q?Firstname_S=C3=BCrname?= <to@example.com>, other@example.com",
|
||
)
|
||
|
||
email = EmailMessage(
|
||
to=['"Sürname, Firstname" <to@example.com>', "other@example.com"],
|
||
)
|
||
self.assertEqual(
|
||
email.message()["To"],
|
||
"=?utf-8?q?S=C3=BCrname=2C_Firstname?= <to@example.com>, other@example.com",
|
||
)
|
||
|
||
def test_unicode_headers(self):
|
||
email = EmailMessage(
|
||
subject="Gżegżółka",
|
||
to=["to@example.com"],
|
||
headers={
|
||
"Sender": '"Firstname Sürname" <sender@example.com>',
|
||
"Comments": "My Sürname is non-ASCII",
|
||
},
|
||
)
|
||
message = email.message()
|
||
self.assertEqual(message["Subject"], "=?utf-8?b?R8W8ZWfFvMOzxYJrYQ==?=")
|
||
self.assertEqual(
|
||
message["Sender"], "=?utf-8?q?Firstname_S=C3=BCrname?= <sender@example.com>"
|
||
)
|
||
self.assertEqual(
|
||
message["Comments"], "=?utf-8?q?My_S=C3=BCrname_is_non-ASCII?="
|
||
)
|
||
|
||
def test_non_utf8_headers_multipart(self):
|
||
"""
|
||
Make sure headers can be set with a different encoding than utf-8 in
|
||
EmailMultiAlternatives as well.
|
||
"""
|
||
headers = {"Date": "Fri, 09 Nov 2001 01:08:47 -0000", "Message-ID": "foo"}
|
||
from_email = "from@example.com"
|
||
to = '"Sürname, Firstname" <to@example.com>'
|
||
text_content = "This is an important message."
|
||
html_content = "<p>This is an <strong>important</strong> message.</p>"
|
||
msg = EmailMultiAlternatives(
|
||
"Message from Firstname Sürname",
|
||
text_content,
|
||
from_email,
|
||
[to],
|
||
headers=headers,
|
||
)
|
||
msg.attach_alternative(html_content, "text/html")
|
||
msg.encoding = "iso-8859-1"
|
||
self.assertEqual(
|
||
msg.message()["To"],
|
||
"=?iso-8859-1?q?S=FCrname=2C_Firstname?= <to@example.com>",
|
||
)
|
||
self.assertEqual(
|
||
msg.message()["Subject"],
|
||
"=?iso-8859-1?q?Message_from_Firstname_S=FCrname?=",
|
||
)
|
||
|
||
def test_multipart_with_attachments(self):
|
||
"""
|
||
EmailMultiAlternatives includes alternatives if the body is empty and
|
||
it has attachments.
|
||
"""
|
||
msg = EmailMultiAlternatives(body="")
|
||
html_content = "<p>This is <strong>html</strong></p>"
|
||
msg.attach_alternative(html_content, "text/html")
|
||
msg.attach("example.txt", "Text file content", "text/plain")
|
||
self.assertIn(html_content, msg.message().as_string())
|
||
|
||
def test_alternatives(self):
|
||
msg = EmailMultiAlternatives()
|
||
html_content = "<p>This is <strong>html</strong></p>"
|
||
mime_type = "text/html"
|
||
msg.attach_alternative(html_content, mime_type)
|
||
|
||
self.assertIsInstance(msg.alternatives[0], EmailAlternative)
|
||
|
||
self.assertEqual(msg.alternatives[0][0], html_content)
|
||
self.assertEqual(msg.alternatives[0].content, html_content)
|
||
|
||
self.assertEqual(msg.alternatives[0][1], mime_type)
|
||
self.assertEqual(msg.alternatives[0].mimetype, mime_type)
|
||
|
||
self.assertIn(html_content, msg.message().as_string())
|
||
|
||
def test_alternatives_constructor(self):
|
||
html_content = "<p>This is <strong>html</strong></p>"
|
||
mime_type = "text/html"
|
||
|
||
msg = EmailMultiAlternatives(
|
||
alternatives=[EmailAlternative(html_content, mime_type)]
|
||
)
|
||
|
||
self.assertIsInstance(msg.alternatives[0], EmailAlternative)
|
||
|
||
self.assertEqual(msg.alternatives[0][0], html_content)
|
||
self.assertEqual(msg.alternatives[0].content, html_content)
|
||
|
||
self.assertEqual(msg.alternatives[0][1], mime_type)
|
||
self.assertEqual(msg.alternatives[0].mimetype, mime_type)
|
||
|
||
self.assertIn(html_content, msg.message().as_string())
|
||
|
||
def test_alternatives_constructor_from_tuple(self):
|
||
html_content = "<p>This is <strong>html</strong></p>"
|
||
mime_type = "text/html"
|
||
|
||
msg = EmailMultiAlternatives(alternatives=[(html_content, mime_type)])
|
||
|
||
self.assertIsInstance(msg.alternatives[0], EmailAlternative)
|
||
|
||
self.assertEqual(msg.alternatives[0][0], html_content)
|
||
self.assertEqual(msg.alternatives[0].content, html_content)
|
||
|
||
self.assertEqual(msg.alternatives[0][1], mime_type)
|
||
self.assertEqual(msg.alternatives[0].mimetype, mime_type)
|
||
|
||
self.assertIn(html_content, msg.message().as_string())
|
||
|
||
def test_none_body(self):
|
||
msg = EmailMessage("subject", None, "from@example.com", ["to@example.com"])
|
||
self.assertEqual(msg.body, "")
|
||
self.assertEqual(msg.message().get_payload(), "")
|
||
|
||
@mock.patch("socket.getfqdn", return_value="漢字")
|
||
def test_non_ascii_dns_non_unicode_email(self, mocked_getfqdn):
|
||
delattr(DNS_NAME, "_fqdn")
|
||
email = EmailMessage()
|
||
email.encoding = "iso-8859-1"
|
||
self.assertIn("@xn--p8s937b>", email.message()["Message-ID"])
|
||
|
||
def test_encoding(self):
|
||
"""
|
||
Regression for #12791 - Encode body correctly with other encodings
|
||
than utf-8
|
||
"""
|
||
email = EmailMessage(body="Firstname Sürname is a great guy.")
|
||
email.encoding = "iso-8859-1"
|
||
message = email.message()
|
||
self.assertMessageHasHeaders(
|
||
message,
|
||
{
|
||
("MIME-Version", "1.0"),
|
||
("Content-Type", 'text/plain; charset="iso-8859-1"'),
|
||
("Content-Transfer-Encoding", "quoted-printable"),
|
||
},
|
||
)
|
||
self.assertEqual(message.get_payload(), "Firstname S=FCrname is a great guy.")
|
||
|
||
def test_encoding_alternatives(self):
|
||
"""
|
||
Encode alternatives correctly with other encodings than utf-8.
|
||
"""
|
||
text_content = "Firstname Sürname is a great guy."
|
||
html_content = "<p>Firstname Sürname is a <strong>great</strong> guy.</p>"
|
||
email = EmailMultiAlternatives(body=text_content)
|
||
email.encoding = "iso-8859-1"
|
||
email.attach_alternative(html_content, "text/html")
|
||
message = email.message()
|
||
# Check the text/plain part.
|
||
payload0 = message.get_payload(0)
|
||
self.assertMessageHasHeaders(
|
||
payload0,
|
||
{
|
||
# (The MIME-Version header is neither required nor meaningful
|
||
# in a subpart, and this check for it can be safely removed.)
|
||
("MIME-Version", "1.0"),
|
||
("Content-Type", 'text/plain; charset="iso-8859-1"'),
|
||
("Content-Transfer-Encoding", "quoted-printable"),
|
||
},
|
||
)
|
||
self.assertEndsWith(
|
||
payload0.as_bytes(), b"\n\nFirstname S=FCrname is a great guy."
|
||
)
|
||
# Check the text/html alternative.
|
||
payload1 = message.get_payload(1)
|
||
self.assertMessageHasHeaders(
|
||
payload1,
|
||
{
|
||
# (The MIME-Version header is neither required nor meaningful
|
||
# in a subpart, and this check for it can be safely removed.)
|
||
("MIME-Version", "1.0"),
|
||
("Content-Type", 'text/html; charset="iso-8859-1"'),
|
||
("Content-Transfer-Encoding", "quoted-printable"),
|
||
},
|
||
)
|
||
self.assertEndsWith(
|
||
payload1.as_bytes(),
|
||
b"\n\n<p>Firstname S=FCrname is a <strong>great</strong> guy.</p>",
|
||
)
|
||
|
||
def test_attachments(self):
|
||
msg = EmailMessage()
|
||
file_name = "example.txt"
|
||
file_content = "Text file content"
|
||
mime_type = "text/plain"
|
||
msg.attach(file_name, file_content, mime_type)
|
||
|
||
self.assertEqual(msg.attachments[0][0], file_name)
|
||
self.assertEqual(msg.attachments[0].filename, file_name)
|
||
|
||
self.assertEqual(msg.attachments[0][1], file_content)
|
||
self.assertEqual(msg.attachments[0].content, file_content)
|
||
|
||
self.assertEqual(msg.attachments[0][2], mime_type)
|
||
self.assertEqual(msg.attachments[0].mimetype, mime_type)
|
||
|
||
attachments = self.get_decoded_attachments(msg)
|
||
self.assertEqual(attachments[0], (file_name, file_content, mime_type))
|
||
|
||
def test_attachments_constructor(self):
|
||
file_name = "example.txt"
|
||
file_content = "Text file content"
|
||
mime_type = "text/plain"
|
||
msg = EmailMessage(
|
||
attachments=[EmailAttachment(file_name, file_content, mime_type)]
|
||
)
|
||
|
||
self.assertIsInstance(msg.attachments[0], EmailAttachment)
|
||
|
||
self.assertEqual(msg.attachments[0][0], file_name)
|
||
self.assertEqual(msg.attachments[0].filename, file_name)
|
||
|
||
self.assertEqual(msg.attachments[0][1], file_content)
|
||
self.assertEqual(msg.attachments[0].content, file_content)
|
||
|
||
self.assertEqual(msg.attachments[0][2], mime_type)
|
||
self.assertEqual(msg.attachments[0].mimetype, mime_type)
|
||
|
||
attachments = self.get_decoded_attachments(msg)
|
||
self.assertEqual(attachments[0], (file_name, file_content, mime_type))
|
||
|
||
def test_attachments_constructor_from_tuple(self):
|
||
file_name = "example.txt"
|
||
file_content = "Text file content"
|
||
mime_type = "text/plain"
|
||
msg = EmailMessage(attachments=[(file_name, file_content, mime_type)])
|
||
|
||
self.assertIsInstance(msg.attachments[0], EmailAttachment)
|
||
|
||
self.assertEqual(msg.attachments[0][0], file_name)
|
||
self.assertEqual(msg.attachments[0].filename, file_name)
|
||
|
||
self.assertEqual(msg.attachments[0][1], file_content)
|
||
self.assertEqual(msg.attachments[0].content, file_content)
|
||
|
||
self.assertEqual(msg.attachments[0][2], mime_type)
|
||
self.assertEqual(msg.attachments[0].mimetype, mime_type)
|
||
|
||
attachments = self.get_decoded_attachments(msg)
|
||
self.assertEqual(attachments[0], (file_name, file_content, mime_type))
|
||
|
||
def test_attachments_constructor_omit_mimetype(self):
|
||
"""
|
||
The mimetype can be omitted from an attachment tuple.
|
||
"""
|
||
msg = EmailMessage(attachments=[("filename1", "content1")])
|
||
filename, content, mimetype = self.get_decoded_attachments(msg)[0]
|
||
self.assertEqual(filename, "filename1")
|
||
self.assertEqual(content, b"content1")
|
||
self.assertEqual(mimetype, "application/octet-stream")
|
||
|
||
def test_attachments_with_alternative_parts(self):
|
||
"""
|
||
Message with attachment and alternative has correct structure (#9367).
|
||
"""
|
||
text_content = "This is an important message."
|
||
html_content = "<p>This is an <strong>important</strong> message.</p>"
|
||
msg = EmailMultiAlternatives(body=text_content)
|
||
msg.attach_alternative(html_content, "text/html")
|
||
msg.attach("an attachment.pdf", b"%PDF-1.4.%...", mimetype="application/pdf")
|
||
msg_bytes = msg.message().as_bytes()
|
||
message = message_from_bytes(msg_bytes)
|
||
self.assertTrue(message.is_multipart())
|
||
self.assertEqual(message.get_content_type(), "multipart/mixed")
|
||
self.assertEqual(message.get_default_type(), "text/plain")
|
||
payload = message.get_payload()
|
||
self.assertEqual(payload[0].get_content_type(), "multipart/alternative")
|
||
self.assertEqual(payload[1].get_content_type(), "application/pdf")
|
||
|
||
def test_decoded_attachments_MIMEText(self):
|
||
txt = MIMEText("content1")
|
||
msg = EmailMessage(attachments=[txt])
|
||
payload = msg.message().get_payload()
|
||
self.assertEqual(payload[0], txt)
|
||
|
||
def test_non_ascii_attachment_filename(self):
|
||
"""Regression test for #14964"""
|
||
msg = EmailMessage(body="Content")
|
||
# Unicode in file name
|
||
msg.attach("une pièce jointe.pdf", b"%PDF-1.4.%...", mimetype="application/pdf")
|
||
attachment = self.get_decoded_attachments(msg)[0]
|
||
self.assertEqual(attachment.filename, "une pièce jointe.pdf")
|
||
|
||
def test_attach_file(self):
|
||
"""
|
||
Test attaching a file against different mimetypes and make sure that
|
||
a file will be attached and sent in some form even if a mismatched
|
||
mimetype is specified.
|
||
"""
|
||
files = (
|
||
# filename, actual mimetype
|
||
("file.txt", "text/plain"),
|
||
("file.png", "image/png"),
|
||
("file_txt", None),
|
||
("file_png", None),
|
||
("file_txt.png", "image/png"),
|
||
("file_png.txt", "text/plain"),
|
||
("file.eml", "message/rfc822"),
|
||
)
|
||
test_mimetypes = ["text/plain", "image/png", None]
|
||
|
||
for basename, real_mimetype in files:
|
||
for mimetype in test_mimetypes:
|
||
with self.subTest(
|
||
basename=basename, real_mimetype=real_mimetype, mimetype=mimetype
|
||
):
|
||
self.assertEqual(mimetypes.guess_type(basename)[0], real_mimetype)
|
||
expected_mimetype = (
|
||
mimetype or real_mimetype or "application/octet-stream"
|
||
)
|
||
file_path = Path(__file__).parent / "attachments" / basename
|
||
expected_content = file_path.read_bytes()
|
||
if expected_mimetype.startswith("text/"):
|
||
try:
|
||
expected_content = expected_content.decode()
|
||
except UnicodeDecodeError:
|
||
expected_mimetype = "application/octet-stream"
|
||
|
||
email = EmailMessage()
|
||
email.attach_file(file_path, mimetype=mimetype)
|
||
|
||
# Check EmailMessage.attachments.
|
||
self.assertEqual(len(email.attachments), 1)
|
||
self.assertEqual(email.attachments[0].filename, basename)
|
||
self.assertEqual(email.attachments[0].mimetype, expected_mimetype)
|
||
self.assertEqual(email.attachments[0].content, expected_content)
|
||
|
||
# Check attachments in the generated message.
|
||
# (The actual content is not checked as variations in platform
|
||
# line endings and rfc822 refolding complicate the logic.)
|
||
attachments = self.get_decoded_attachments(email)
|
||
self.assertEqual(len(attachments), 1)
|
||
actual = attachments[0]
|
||
self.assertEqual(actual.filename, basename)
|
||
self.assertEqual(actual.mimetype, expected_mimetype)
|
||
|
||
def test_attach_text_as_bytes(self):
|
||
msg = EmailMessage()
|
||
msg.attach("file.txt", b"file content")
|
||
filename, content, mimetype = self.get_decoded_attachments(msg)[0]
|
||
self.assertEqual(filename, "file.txt")
|
||
self.assertEqual(content, "file content") # (decoded)
|
||
self.assertEqual(mimetype, "text/plain")
|
||
|
||
def test_attach_utf8_text_as_bytes(self):
|
||
"""
|
||
Non-ASCII characters encoded as valid UTF-8 are correctly transported
|
||
in a form that can be decoded at the receiving end.
|
||
"""
|
||
msg = EmailMessage()
|
||
msg.attach("file.txt", b"\xc3\xa4") # UTF-8 encoded a umlaut.
|
||
filename, content, mimetype = self.get_decoded_attachments(msg)[0]
|
||
self.assertEqual(filename, "file.txt")
|
||
self.assertEqual(content, "ä") # (decoded)
|
||
self.assertEqual(mimetype, "text/plain")
|
||
|
||
def test_attach_non_utf8_text_as_bytes(self):
|
||
"""
|
||
Binary data that can't be decoded as UTF-8 overrides the MIME type
|
||
instead of decoding the data.
|
||
"""
|
||
msg = EmailMessage()
|
||
msg.attach("file.txt", b"\xff") # Invalid UTF-8.
|
||
filename, content, mimetype = self.get_decoded_attachments(msg)[0]
|
||
self.assertEqual(filename, "file.txt")
|
||
# Content should be passed through unmodified.
|
||
self.assertEqual(content, b"\xff")
|
||
self.assertEqual(mimetype, "application/octet-stream")
|
||
|
||
def test_attach_mime_image(self):
|
||
"""
|
||
EmailMessage.attach() docs: "You can pass it
|
||
a single argument that is a MIMEBase instance."
|
||
"""
|
||
# This also verifies complex attachments with extra header fields.
|
||
email = EmailMessage()
|
||
image = MIMEImage(b"GIF89a...", "gif")
|
||
image["Content-Disposition"] = "inline"
|
||
image["Content-ID"] = "<content-id@example.org>"
|
||
email.attach(image)
|
||
|
||
attachments = self.get_raw_attachments(email)
|
||
self.assertEqual(len(attachments), 1)
|
||
image_att = attachments[0]
|
||
self.assertEqual(image_att.get_content_type(), "image/gif")
|
||
self.assertEqual(image_att.get_content_disposition(), "inline")
|
||
self.assertEqual(image_att["Content-ID"], "<content-id@example.org>")
|
||
self.assertEqual(image_att.get_content(), b"GIF89a...")
|
||
self.assertIsNone(image_att.get_filename())
|
||
|
||
def test_attach_mime_image_in_constructor(self):
|
||
image = MIMEImage(b"\x89PNG...", "png")
|
||
image["Content-Disposition"] = "attachment; filename=test.png"
|
||
email = EmailMessage(attachments=[image])
|
||
|
||
attachments = self.get_raw_attachments(email)
|
||
self.assertEqual(len(attachments), 1)
|
||
image_att = attachments[0]
|
||
self.assertEqual(image_att.get_content_type(), "image/png")
|
||
self.assertEqual(image_att.get_content(), b"\x89PNG...")
|
||
self.assertEqual(image_att.get_content_disposition(), "attachment")
|
||
self.assertEqual(image_att.get_filename(), "test.png")
|
||
|
||
def test_attach_rfc822_message(self):
|
||
"""
|
||
EmailMessage.attach() docs: "If you specify a mimetype of message/rfc822,
|
||
it will also accept django.core.mail.EmailMessage and email.message.Message."
|
||
"""
|
||
# django.core.mail.EmailMessage
|
||
django_email = EmailMessage("child subject", "child body")
|
||
# email.message.Message
|
||
py_message = PyMessage()
|
||
py_message["Subject"] = "child subject"
|
||
py_message.set_payload("child body")
|
||
# email.message.EmailMessage
|
||
py_email_message = PyEmailMessage()
|
||
py_email_message["Subject"] = "child subject"
|
||
py_email_message.set_content("child body")
|
||
|
||
cases = [
|
||
django_email,
|
||
py_message,
|
||
py_email_message,
|
||
# Should also allow message serialized as str or bytes.
|
||
py_message.as_string(),
|
||
py_message.as_bytes(),
|
||
]
|
||
|
||
for child_message in cases:
|
||
with self.subTest(child_type=child_message.__class__):
|
||
email = EmailMessage("parent message", "parent body")
|
||
email.attach(content=child_message, mimetype="message/rfc822")
|
||
self.assertEqual(len(email.attachments), 1)
|
||
self.assertIsInstance(email.attachments[0], EmailAttachment)
|
||
self.assertEqual(email.attachments[0].mimetype, "message/rfc822")
|
||
|
||
# Make sure it is serialized correctly: a message/rfc822 attachment
|
||
# whose "body" content (payload) is the "encapsulated" (child) message.
|
||
attachments = self.get_raw_attachments(email)
|
||
self.assertEqual(len(attachments), 1)
|
||
rfc822_attachment = attachments[0]
|
||
self.assertEqual(rfc822_attachment.get_content_type(), "message/rfc822")
|
||
|
||
attached_message = rfc822_attachment.get_content()
|
||
self.assertEqual(attached_message["Subject"], "child subject")
|
||
self.assertEqual(attached_message.get_content().rstrip(), "child body")
|
||
|
||
# Regression for #18967: Per RFC 2046 5.2.1, "No encoding other
|
||
# than '7bit', '8bit', or 'binary' is permitted for the body of
|
||
# a 'message/rfc822' entity." (Default CTE is "7bit".)
|
||
cte = rfc822_attachment.get("Content-Transfer-Encoding", "7bit")
|
||
self.assertIn(cte, ("7bit", "8bit", "binary"))
|
||
|
||
# Any properly declared CTE is allowed for the attached message itself
|
||
# (including quoted-printable or base64). For the plain ASCII content
|
||
# in this test, we'd expect 7bit.
|
||
child_cte = attached_message.get("Content-Transfer-Encoding", "7bit")
|
||
self.assertEqual(child_cte, "7bit")
|
||
self.assertEqual(attached_message.get_content_type(), "text/plain")
|
||
|
||
def test_attach_mimebase_prohibits_other_params(self):
|
||
email_msg = EmailMessage()
|
||
txt = MIMEText("content")
|
||
msg = (
|
||
"content and mimetype must not be given when a MIMEBase instance "
|
||
"is provided."
|
||
)
|
||
with self.assertRaisesMessage(ValueError, msg):
|
||
email_msg.attach(txt, content="content")
|
||
with self.assertRaisesMessage(ValueError, msg):
|
||
email_msg.attach(txt, mimetype="text/plain")
|
||
|
||
def test_attach_content_is_required(self):
|
||
email_msg = EmailMessage()
|
||
msg = "content must be provided."
|
||
with self.assertRaisesMessage(ValueError, msg):
|
||
email_msg.attach("file.txt", mimetype="application/pdf")
|
||
|
||
def test_dummy_backend(self):
|
||
"""
|
||
Make sure that dummy backends returns correct number of sent messages
|
||
"""
|
||
connection = dummy.EmailBackend()
|
||
email = EmailMessage(to=["to@example.com"])
|
||
self.assertEqual(connection.send_messages([email, email, email]), 3)
|
||
|
||
def test_arbitrary_keyword(self):
|
||
"""
|
||
Make sure that get_connection() accepts arbitrary keyword that might be
|
||
used with custom backends.
|
||
"""
|
||
c = mail.get_connection(fail_silently=True, foo="bar")
|
||
self.assertTrue(c.fail_silently)
|
||
|
||
def test_custom_backend(self):
|
||
"""Test custom backend defined in this suite."""
|
||
conn = mail.get_connection("mail.custombackend.EmailBackend")
|
||
self.assertTrue(hasattr(conn, "test_outbox"))
|
||
email = EmailMessage(to=["to@example.com"])
|
||
conn.send_messages([email])
|
||
self.assertEqual(len(conn.test_outbox), 1)
|
||
|
||
def test_backend_arg(self):
|
||
"""Test backend argument of mail.get_connection()"""
|
||
self.assertIsInstance(
|
||
mail.get_connection("django.core.mail.backends.smtp.EmailBackend"),
|
||
smtp.EmailBackend,
|
||
)
|
||
self.assertIsInstance(
|
||
mail.get_connection("django.core.mail.backends.locmem.EmailBackend"),
|
||
locmem.EmailBackend,
|
||
)
|
||
self.assertIsInstance(
|
||
mail.get_connection("django.core.mail.backends.dummy.EmailBackend"),
|
||
dummy.EmailBackend,
|
||
)
|
||
self.assertIsInstance(
|
||
mail.get_connection("django.core.mail.backends.console.EmailBackend"),
|
||
console.EmailBackend,
|
||
)
|
||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||
self.assertIsInstance(
|
||
mail.get_connection(
|
||
"django.core.mail.backends.filebased.EmailBackend",
|
||
file_path=tmp_dir,
|
||
),
|
||
filebased.EmailBackend,
|
||
)
|
||
|
||
msg = " not object"
|
||
with self.assertRaisesMessage(TypeError, msg):
|
||
mail.get_connection(
|
||
"django.core.mail.backends.filebased.EmailBackend", file_path=object()
|
||
)
|
||
self.assertIsInstance(mail.get_connection(), locmem.EmailBackend)
|
||
|
||
@override_settings(
|
||
EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend",
|
||
ADMINS=[("nobody", "nobody@example.com")],
|
||
MANAGERS=[("nobody", "nobody@example.com")],
|
||
)
|
||
def test_connection_arg(self):
|
||
"""Test connection argument to send_mail(), et. al."""
|
||
mail.outbox = []
|
||
|
||
# Send using non-default connection
|
||
connection = mail.get_connection("mail.custombackend.EmailBackend")
|
||
send_mail(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com"],
|
||
connection=connection,
|
||
)
|
||
self.assertEqual(mail.outbox, [])
|
||
self.assertEqual(len(connection.test_outbox), 1)
|
||
self.assertEqual(connection.test_outbox[0].subject, "Subject")
|
||
|
||
connection = mail.get_connection("mail.custombackend.EmailBackend")
|
||
send_mass_mail(
|
||
[
|
||
("Subject1", "Content1", "from1@example.com", ["to1@example.com"]),
|
||
("Subject2", "Content2", "from2@example.com", ["to2@example.com"]),
|
||
],
|
||
connection=connection,
|
||
)
|
||
self.assertEqual(mail.outbox, [])
|
||
self.assertEqual(len(connection.test_outbox), 2)
|
||
self.assertEqual(connection.test_outbox[0].subject, "Subject1")
|
||
self.assertEqual(connection.test_outbox[1].subject, "Subject2")
|
||
|
||
connection = mail.get_connection("mail.custombackend.EmailBackend")
|
||
mail_admins("Admin message", "Content", connection=connection)
|
||
self.assertEqual(mail.outbox, [])
|
||
self.assertEqual(len(connection.test_outbox), 1)
|
||
self.assertEqual(connection.test_outbox[0].subject, "[Django] Admin message")
|
||
|
||
connection = mail.get_connection("mail.custombackend.EmailBackend")
|
||
mail_managers("Manager message", "Content", connection=connection)
|
||
self.assertEqual(mail.outbox, [])
|
||
self.assertEqual(len(connection.test_outbox), 1)
|
||
self.assertEqual(connection.test_outbox[0].subject, "[Django] Manager message")
|
||
|
||
def test_dont_mangle_from_in_body(self):
|
||
# Regression for #13433 - Make sure that EmailMessage doesn't mangle
|
||
# 'From ' in message body.
|
||
email = EmailMessage(body="From the future")
|
||
self.assertNotIn(b">From the future", email.message().as_bytes())
|
||
|
||
def test_body_content_transfer_encoding(self):
|
||
# Shouldn't use base64 or quoted-printable, instead should detect it
|
||
# can represent content with 7-bit data (#3472, #11212).
|
||
msg = EmailMessage(body="Body with only ASCII characters.")
|
||
s = msg.message().as_bytes()
|
||
self.assertIn(b"Content-Transfer-Encoding: 7bit", s)
|
||
|
||
# Shouldn't use base64 or quoted-printable, instead should detect
|
||
# it can represent content with 8-bit data.
|
||
msg = EmailMessage(body="Body with latin characters: àáä.")
|
||
s = msg.message().as_bytes()
|
||
self.assertIn(b"Content-Transfer-Encoding: 8bit", s)
|
||
# The following test is left over from Python 2 and can be safely removed.
|
||
# 8bit CTE within a Unicode str is not meaningful, and Python's modern
|
||
# email api won't generate it. (The test still works with the legacy api.)
|
||
s = msg.message().as_string()
|
||
self.assertIn("Content-Transfer-Encoding: 8bit", s)
|
||
|
||
# Long body lines that require folding should use quoted-printable or base64,
|
||
# whichever is shorter. However, Python's legacy email API avoids re-folding
|
||
# non-ASCII text and just uses CTE 8bit. (The modern API would correctly choose
|
||
# base64 here. Any of these is deliverable.)
|
||
msg = EmailMessage(
|
||
"Subject",
|
||
"Body with non latin characters: А Б В Г Д Е Ж Ѕ З И І К Л М Н О П.",
|
||
"bounce@example.com",
|
||
["to@example.com"],
|
||
headers={"From": "from@example.com"},
|
||
)
|
||
s = msg.message().as_bytes()
|
||
self.assertIn(b"Content-Transfer-Encoding: 8bit", s)
|
||
# The following test is left over from Python 2.
|
||
s = msg.message().as_string()
|
||
self.assertIn("Content-Transfer-Encoding: 8bit", s)
|
||
|
||
def test_custom_utf8_encoding(self):
|
||
"""A UTF-8 charset with a custom body encoding is respected."""
|
||
body = "Body with latin characters: àáä."
|
||
msg = EmailMessage("Subject", body, "bounce@example.com", ["to@example.com"])
|
||
encoding = charset.Charset("utf-8")
|
||
encoding.body_encoding = charset.QP
|
||
msg.encoding = encoding
|
||
message = msg.message()
|
||
self.assertMessageHasHeaders(
|
||
message,
|
||
{
|
||
("MIME-Version", "1.0"),
|
||
("Content-Type", 'text/plain; charset="utf-8"'),
|
||
("Content-Transfer-Encoding", "quoted-printable"),
|
||
},
|
||
)
|
||
self.assertEqual(message.get_payload(), encoding.body_encode(body))
|
||
|
||
def test_sanitize_address(self):
|
||
"""Email addresses are properly sanitized."""
|
||
for email_address, encoding, expected_result in (
|
||
# ASCII addresses.
|
||
("to@example.com", "ascii", "to@example.com"),
|
||
("to@example.com", "utf-8", "to@example.com"),
|
||
(("A name", "to@example.com"), "ascii", "A name <to@example.com>"),
|
||
(
|
||
("A name", "to@example.com"),
|
||
"utf-8",
|
||
"A name <to@example.com>",
|
||
),
|
||
("localpartonly", "ascii", "localpartonly"),
|
||
# ASCII addresses with display names.
|
||
("A name <to@example.com>", "ascii", "A name <to@example.com>"),
|
||
("A name <to@example.com>", "utf-8", "A name <to@example.com>"),
|
||
('"A name" <to@example.com>', "ascii", "A name <to@example.com>"),
|
||
('"A name" <to@example.com>', "utf-8", "A name <to@example.com>"),
|
||
# Unicode addresses: IDNA encoded domain supported per RFC-5890.
|
||
("to@éxample.com", "utf-8", "to@xn--xample-9ua.com"),
|
||
# The next three cases should be removed when fixing #35713.
|
||
# (An 'encoded-word' localpart is prohibited by RFC-2047, and not
|
||
# supported by any known mail service.)
|
||
("tó@example.com", "utf-8", "=?utf-8?b?dMOz?=@example.com"),
|
||
(
|
||
("Tó Example", "tó@example.com"),
|
||
"utf-8",
|
||
"=?utf-8?q?T=C3=B3_Example?= <=?utf-8?b?dMOz?=@example.com>",
|
||
),
|
||
(
|
||
"Tó Example <tó@example.com>",
|
||
"utf-8",
|
||
"=?utf-8?q?T=C3=B3_Example?= <=?utf-8?b?dMOz?=@example.com>",
|
||
),
|
||
# IDNA addresses with display names.
|
||
(
|
||
"To Example <to@éxample.com>",
|
||
"ascii",
|
||
"To Example <to@xn--xample-9ua.com>",
|
||
),
|
||
(
|
||
"To Example <to@éxample.com>",
|
||
"utf-8",
|
||
"To Example <to@xn--xample-9ua.com>",
|
||
),
|
||
# Addresses with two @ signs.
|
||
('"to@other.com"@example.com', "utf-8", r'"to@other.com"@example.com'),
|
||
(
|
||
'"to@other.com" <to@example.com>',
|
||
"utf-8",
|
||
'"to@other.com" <to@example.com>',
|
||
),
|
||
(
|
||
("To Example", "to@other.com@example.com"),
|
||
"utf-8",
|
||
'To Example <"to@other.com"@example.com>',
|
||
),
|
||
# Addresses with long unicode display names.
|
||
(
|
||
"Tó Example very long" * 4 + " <to@example.com>",
|
||
"utf-8",
|
||
"=?utf-8?q?T=C3=B3_Example_very_longT=C3=B3_Example_very_longT"
|
||
"=C3=B3_Example_?=\n"
|
||
" =?utf-8?q?very_longT=C3=B3_Example_very_long?= "
|
||
"<to@example.com>",
|
||
),
|
||
(
|
||
("Tó Example very long" * 4, "to@example.com"),
|
||
"utf-8",
|
||
"=?utf-8?q?T=C3=B3_Example_very_longT=C3=B3_Example_very_longT"
|
||
"=C3=B3_Example_?=\n"
|
||
" =?utf-8?q?very_longT=C3=B3_Example_very_long?= "
|
||
"<to@example.com>",
|
||
),
|
||
# Address with long display name and unicode domain.
|
||
(
|
||
("To Example very long" * 4, "to@exampl€.com"),
|
||
"utf-8",
|
||
"To Example very longTo Example very longTo Example very longT"
|
||
"o Example very\n"
|
||
" long <to@xn--exampl-nc1c.com>",
|
||
),
|
||
):
|
||
with self.subTest(email_address=email_address, encoding=encoding):
|
||
self.assertEqual(
|
||
sanitize_address(email_address, encoding), expected_result
|
||
)
|
||
|
||
def test_sanitize_address_invalid(self):
|
||
for email_address in (
|
||
# Invalid address with two @ signs.
|
||
"to@other.com@example.com",
|
||
# Invalid address without the quotes.
|
||
"to@other.com <to@example.com>",
|
||
# Other invalid addresses.
|
||
"@",
|
||
"to@",
|
||
"@example.com",
|
||
("", ""),
|
||
):
|
||
with self.subTest(email_address=email_address):
|
||
with self.assertRaisesMessage(ValueError, "Invalid address"):
|
||
sanitize_address(email_address, encoding="utf-8")
|
||
|
||
def test_sanitize_address_header_injection(self):
|
||
msg = "Invalid address; address parts cannot contain newlines."
|
||
tests = [
|
||
"Name\nInjection <to@example.com>",
|
||
("Name\nInjection", "to@xample.com"),
|
||
"Name <to\ninjection@example.com>",
|
||
("Name", "to\ninjection@example.com"),
|
||
]
|
||
for email_address in tests:
|
||
with self.subTest(email_address=email_address):
|
||
with self.assertRaisesMessage(ValueError, msg):
|
||
sanitize_address(email_address, encoding="utf-8")
|
||
|
||
def test_email_multi_alternatives_content_mimetype_none(self):
|
||
email_msg = EmailMultiAlternatives()
|
||
msg = "Both content and mimetype must be provided."
|
||
with self.assertRaisesMessage(ValueError, msg):
|
||
email_msg.attach_alternative(None, "text/html")
|
||
with self.assertRaisesMessage(ValueError, msg):
|
||
email_msg.attach_alternative("<p>content</p>", None)
|
||
|
||
def test_mime_structure(self):
|
||
"""
|
||
Check generated messages have the expected MIME parts and nesting.
|
||
"""
|
||
html_body = EmailAlternative("<p>HTML</p>", "text/html")
|
||
image = EmailAttachment("image.gif", b"\x89PNG...", "image/png")
|
||
rfc822_attachment = EmailAttachment(
|
||
None, EmailMessage(body="text"), "message/rfc822"
|
||
)
|
||
cases = [
|
||
# name, email (EmailMessage or subclass), expected structure
|
||
(
|
||
"single body",
|
||
EmailMessage(body="text"),
|
||
"""
|
||
text/plain
|
||
""",
|
||
),
|
||
(
|
||
"single body with attachment",
|
||
EmailMessage(body="text", attachments=[image]),
|
||
"""
|
||
multipart/mixed
|
||
text/plain
|
||
image/png
|
||
""",
|
||
),
|
||
(
|
||
"alternative bodies",
|
||
EmailMultiAlternatives(body="text", alternatives=[html_body]),
|
||
"""
|
||
multipart/alternative
|
||
text/plain
|
||
text/html
|
||
""",
|
||
),
|
||
(
|
||
"alternative bodies with attachments",
|
||
EmailMultiAlternatives(
|
||
body="text", alternatives=[html_body], attachments=[image]
|
||
),
|
||
"""
|
||
multipart/mixed
|
||
multipart/alternative
|
||
text/plain
|
||
text/html
|
||
image/png
|
||
""",
|
||
),
|
||
(
|
||
"alternative bodies with rfc822 attachment",
|
||
EmailMultiAlternatives(
|
||
body="text",
|
||
alternatives=[html_body],
|
||
attachments=[rfc822_attachment],
|
||
),
|
||
"""
|
||
multipart/mixed
|
||
multipart/alternative
|
||
text/plain
|
||
text/html
|
||
message/rfc822
|
||
text/plain
|
||
""",
|
||
),
|
||
(
|
||
"attachment only",
|
||
EmailMessage(attachments=[image]),
|
||
# Avoid empty text/plain body.
|
||
"""
|
||
multipart/mixed
|
||
image/png
|
||
""",
|
||
),
|
||
(
|
||
"alternative only",
|
||
EmailMultiAlternatives(alternatives=[html_body]),
|
||
# Avoid empty text/plain body.
|
||
"""
|
||
multipart/alternative
|
||
text/html
|
||
""",
|
||
),
|
||
(
|
||
"alternative and attachment only",
|
||
EmailMultiAlternatives(alternatives=[html_body], attachments=[image]),
|
||
"""
|
||
multipart/mixed
|
||
multipart/alternative
|
||
text/html
|
||
image/png
|
||
""",
|
||
),
|
||
(
|
||
"empty EmailMessage",
|
||
EmailMessage(),
|
||
"""
|
||
text/plain
|
||
""",
|
||
),
|
||
(
|
||
"empty EmailMultiAlternatives",
|
||
EmailMultiAlternatives(),
|
||
"""
|
||
text/plain
|
||
""",
|
||
),
|
||
]
|
||
for name, email, expected in cases:
|
||
expected = dedent(expected).lstrip()
|
||
with self.subTest(name=name):
|
||
message = email.message()
|
||
structure = self.get_message_structure(message)
|
||
self.assertEqual(structure, expected)
|
||
|
||
def test_body_contains(self):
|
||
email_msg = EmailMultiAlternatives()
|
||
email_msg.body = "I am content."
|
||
self.assertIs(email_msg.body_contains("I am"), True)
|
||
self.assertIs(email_msg.body_contains("I am content."), True)
|
||
|
||
email_msg.attach_alternative("<p>I am different content.</p>", "text/html")
|
||
self.assertIs(email_msg.body_contains("I am"), True)
|
||
self.assertIs(email_msg.body_contains("I am content."), False)
|
||
self.assertIs(email_msg.body_contains("<p>I am different content.</p>"), False)
|
||
|
||
def test_body_contains_alternative_non_text(self):
|
||
email_msg = EmailMultiAlternatives()
|
||
email_msg.body = "I am content."
|
||
email_msg.attach_alternative("I am content.", "text/html")
|
||
email_msg.attach_alternative(b"I am a song.", "audio/mpeg")
|
||
self.assertIs(email_msg.body_contains("I am content"), True)
|
||
|
||
def test_all_params_optional(self):
|
||
"""
|
||
EmailMessage class docs: "All parameters are optional"
|
||
"""
|
||
email = EmailMessage()
|
||
self.assertIsInstance(email.message(), PyMessage) # force serialization.
|
||
|
||
email = EmailMultiAlternatives()
|
||
self.assertIsInstance(email.message(), PyMessage) # force serialization.
|
||
|
||
def test_positional_arguments_order(self):
|
||
"""
|
||
EmailMessage class docs: "… is initialized with the following parameters
|
||
(in the given order, if positional arguments are used)."
|
||
"""
|
||
connection = mail.get_connection()
|
||
email = EmailMessage(
|
||
# (If you need to insert/remove/reorder any params here,
|
||
# that indicates a breaking change to documented behavior.)
|
||
"subject",
|
||
"body",
|
||
"from@example.com",
|
||
["to@example.com"],
|
||
["bcc@example.com"],
|
||
connection,
|
||
[EmailAttachment("file.txt", "attachment", "text/plain")],
|
||
{"X-Header": "custom header"},
|
||
["cc@example.com"],
|
||
["reply-to@example.com"],
|
||
# (New options can be added below here, ideally as keyword-only args.)
|
||
)
|
||
|
||
message = email.message()
|
||
self.assertEqual(message.get_all("Subject"), ["subject"])
|
||
self.assertEqual(message.get_all("From"), ["from@example.com"])
|
||
self.assertEqual(message.get_all("To"), ["to@example.com"])
|
||
self.assertEqual(message.get_all("X-Header"), ["custom header"])
|
||
self.assertEqual(message.get_all("Cc"), ["cc@example.com"])
|
||
self.assertEqual(message.get_all("Reply-To"), ["reply-to@example.com"])
|
||
self.assertEqual(message.get_payload(0).get_payload(), "body")
|
||
self.assertEqual(
|
||
self.get_decoded_attachments(email),
|
||
[("file.txt", "attachment", "text/plain")],
|
||
)
|
||
self.assertEqual(
|
||
email.recipients(), ["to@example.com", "cc@example.com", "bcc@example.com"]
|
||
)
|
||
self.assertIs(email.get_connection(), connection)
|
||
|
||
def test_all_params_can_be_set_before_send(self):
|
||
"""
|
||
EmailMessage class docs: "All parameters … can be set at any time
|
||
prior to calling the send() method."
|
||
"""
|
||
# This is meant to verify EmailMessage.__init__() doesn't apply any
|
||
# special processing that would be missing for properties set later.
|
||
original_connection = mail.get_connection(username="original")
|
||
new_connection = mail.get_connection(username="new")
|
||
email = EmailMessage(
|
||
"original subject",
|
||
"original body",
|
||
"original-from@example.com",
|
||
["original-to@example.com"],
|
||
["original-bcc@example.com"],
|
||
original_connection,
|
||
[EmailAttachment("original.txt", "original attachment", "text/plain")],
|
||
{"X-Header": "original header"},
|
||
["original-cc@example.com"],
|
||
["original-reply-to@example.com"],
|
||
)
|
||
email.subject = "new subject"
|
||
email.body = "new body"
|
||
email.from_email = "new-from@example.com"
|
||
email.to = ["new-to@example.com"]
|
||
email.bcc = ["new-bcc@example.com"]
|
||
email.connection = new_connection
|
||
email.attachments = [
|
||
("new1.txt", "new attachment 1", "text/plain"), # plain tuple.
|
||
EmailAttachment("new2.txt", "new attachment 2", "text/csv"),
|
||
MIMEImage(b"GIF89a...", "gif"),
|
||
]
|
||
email.extra_headers = {"X-Header": "new header"}
|
||
email.cc = ["new-cc@example.com"]
|
||
email.reply_to = ["new-reply-to@example.com"]
|
||
|
||
message = email.message()
|
||
self.assertEqual(message.get_all("Subject"), ["new subject"])
|
||
self.assertEqual(message.get_all("From"), ["new-from@example.com"])
|
||
self.assertEqual(message.get_all("To"), ["new-to@example.com"])
|
||
self.assertEqual(message.get_all("X-Header"), ["new header"])
|
||
self.assertEqual(message.get_all("Cc"), ["new-cc@example.com"])
|
||
self.assertEqual(message.get_all("Reply-To"), ["new-reply-to@example.com"])
|
||
self.assertEqual(message.get_payload(0).get_payload(), "new body")
|
||
self.assertEqual(
|
||
self.get_decoded_attachments(email),
|
||
[
|
||
("new1.txt", "new attachment 1", "text/plain"),
|
||
("new2.txt", "new attachment 2", "text/csv"),
|
||
(None, b"GIF89a...", "image/gif"),
|
||
],
|
||
)
|
||
self.assertEqual(
|
||
email.recipients(),
|
||
["new-to@example.com", "new-cc@example.com", "new-bcc@example.com"],
|
||
)
|
||
self.assertIs(email.get_connection(), new_connection)
|
||
self.assertNotIn("original", message.as_string())
|
||
|
||
|
||
@requires_tz_support
|
||
class MailTimeZoneTests(MailTestsMixin, SimpleTestCase):
|
||
@override_settings(
|
||
EMAIL_USE_LOCALTIME=False, USE_TZ=True, TIME_ZONE="Africa/Algiers"
|
||
)
|
||
def test_date_header_utc(self):
|
||
"""
|
||
EMAIL_USE_LOCALTIME=False creates a datetime in UTC.
|
||
"""
|
||
email = EmailMessage()
|
||
self.assertEndsWith(email.message()["Date"], "-0000")
|
||
|
||
@override_settings(
|
||
EMAIL_USE_LOCALTIME=True, USE_TZ=True, TIME_ZONE="Africa/Algiers"
|
||
)
|
||
def test_date_header_localtime(self):
|
||
"""
|
||
EMAIL_USE_LOCALTIME=True creates a datetime in the local time zone.
|
||
"""
|
||
email = EmailMessage()
|
||
# Africa/Algiers is UTC+1 year round.
|
||
self.assertEndsWith(email.message()["Date"], "+0100")
|
||
|
||
|
||
class PythonGlobalState(SimpleTestCase):
|
||
"""
|
||
Tests for #12422 -- Django smarts (#2472/#11212) with charset of utf-8 text
|
||
parts shouldn't pollute global email Python package charset registry when
|
||
django.mail.message is imported.
|
||
"""
|
||
|
||
def test_utf8(self):
|
||
txt = MIMEText("UTF-8 encoded body", "plain", "utf-8")
|
||
self.assertIn("Content-Transfer-Encoding: base64", txt.as_string())
|
||
|
||
def test_7bit(self):
|
||
txt = MIMEText("Body with only ASCII characters.", "plain", "utf-8")
|
||
self.assertIn("Content-Transfer-Encoding: base64", txt.as_string())
|
||
|
||
def test_8bit_latin(self):
|
||
txt = MIMEText("Body with latin characters: àáä.", "plain", "utf-8")
|
||
self.assertIn("Content-Transfer-Encoding: base64", txt.as_string())
|
||
|
||
def test_8bit_non_latin(self):
|
||
txt = MIMEText(
|
||
"Body with non latin characters: А Б В Г Д Е Ж Ѕ З И І К Л М Н О П.",
|
||
"plain",
|
||
"utf-8",
|
||
)
|
||
self.assertIn("Content-Transfer-Encoding: base64", txt.as_string())
|
||
|
||
|
||
class BaseEmailBackendTests(MailTestsMixin):
|
||
email_backend = None
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.enterClassContext(override_settings(EMAIL_BACKEND=cls.email_backend))
|
||
super().setUpClass()
|
||
|
||
def get_mailbox_content(self):
|
||
raise NotImplementedError(
|
||
"subclasses of BaseEmailBackendTests must provide a get_mailbox_content() "
|
||
"method"
|
||
)
|
||
|
||
def flush_mailbox(self):
|
||
raise NotImplementedError(
|
||
"subclasses of BaseEmailBackendTests may require a flush_mailbox() method"
|
||
)
|
||
|
||
def get_the_message(self):
|
||
mailbox = self.get_mailbox_content()
|
||
self.assertEqual(
|
||
len(mailbox),
|
||
1,
|
||
"Expected exactly one message, got %d.\n%r"
|
||
% (len(mailbox), [m.as_string() for m in mailbox]),
|
||
)
|
||
return mailbox[0]
|
||
|
||
def test_send(self):
|
||
email = EmailMessage(
|
||
"Subject", "Content", "from@example.com", ["to@example.com"]
|
||
)
|
||
num_sent = mail.get_connection().send_messages([email])
|
||
self.assertEqual(num_sent, 1)
|
||
message = self.get_the_message()
|
||
self.assertEqual(message["subject"], "Subject")
|
||
self.assertEqual(message.get_payload(), "Content")
|
||
self.assertEqual(message["from"], "from@example.com")
|
||
self.assertEqual(message.get_all("to"), ["to@example.com"])
|
||
|
||
def test_send_unicode(self):
|
||
email = EmailMessage(
|
||
"Chère maman", "Je t'aime très fort", "from@example.com", ["to@example.com"]
|
||
)
|
||
num_sent = mail.get_connection().send_messages([email])
|
||
self.assertEqual(num_sent, 1)
|
||
message = self.get_the_message()
|
||
self.assertEqual(message["subject"], "Chère maman")
|
||
self.assertIn(b"Subject: =?utf-8?q?Ch=C3=A8re_maman?=", message.as_bytes())
|
||
self.assertEqual(message.get_content(), "Je t'aime très fort")
|
||
|
||
def test_send_long_lines(self):
|
||
"""
|
||
Email line length is limited to 998 chars by the RFC 5322 Section
|
||
2.1.1.
|
||
Message body containing longer lines are converted to Quoted-Printable
|
||
to avoid having to insert newlines, which could be hairy to do properly.
|
||
"""
|
||
# Unencoded body length is < 998 (840) but > 998 when utf-8 encoded.
|
||
email = EmailMessage(
|
||
"Subject", "В южных морях " * 60, "from@example.com", ["to@example.com"]
|
||
)
|
||
email.send()
|
||
message = self.get_the_message()
|
||
self.assertMessageHasHeaders(
|
||
message,
|
||
{
|
||
("MIME-Version", "1.0"),
|
||
("Content-Type", 'text/plain; charset="utf-8"'),
|
||
("Content-Transfer-Encoding", "quoted-printable"),
|
||
},
|
||
)
|
||
|
||
def test_send_many(self):
|
||
email1 = EmailMessage(to=["to-1@example.com"])
|
||
email2 = EmailMessage(to=["to-2@example.com"])
|
||
# send_messages() may take a list or an iterator.
|
||
emails_lists = ([email1, email2], iter((email1, email2)))
|
||
for emails_list in emails_lists:
|
||
with self.subTest(emails_list=repr(emails_list)):
|
||
num_sent = mail.get_connection().send_messages(emails_list)
|
||
self.assertEqual(num_sent, 2)
|
||
messages = self.get_mailbox_content()
|
||
self.assertEqual(len(messages), 2)
|
||
self.assertEqual(messages[0]["To"], "to-1@example.com")
|
||
self.assertEqual(messages[1]["To"], "to-2@example.com")
|
||
self.flush_mailbox()
|
||
|
||
def test_send_verbose_name(self):
|
||
email = EmailMessage(
|
||
from_email='"Firstname Sürname" <from@example.com>',
|
||
to=["to@example.com"],
|
||
)
|
||
email.send()
|
||
message = self.get_the_message()
|
||
self.assertEqual(message["from"], "Firstname Sürname <from@example.com>")
|
||
self.assertIn(
|
||
b"From: =?utf-8?q?Firstname_S=C3=BCrname?= <from@example.com>",
|
||
message.as_bytes(),
|
||
)
|
||
|
||
def test_plaintext_send_mail(self):
|
||
"""
|
||
Test send_mail without the html_message
|
||
regression test for adding html_message parameter to send_mail()
|
||
"""
|
||
send_mail("Subject", "Content", "sender@example.com", ["nobody@example.com"])
|
||
message = self.get_the_message()
|
||
|
||
self.assertEqual(message.get("subject"), "Subject")
|
||
self.assertEqual(message.get_all("to"), ["nobody@example.com"])
|
||
self.assertFalse(message.is_multipart())
|
||
self.assertEqual(message.get_payload(), "Content")
|
||
self.assertEqual(message.get_content_type(), "text/plain")
|
||
|
||
def test_html_send_mail(self):
|
||
"""Test html_message argument to send_mail"""
|
||
send_mail(
|
||
"Subject",
|
||
"Content",
|
||
"sender@example.com",
|
||
["nobody@example.com"],
|
||
html_message="HTML Content",
|
||
)
|
||
message = self.get_the_message()
|
||
|
||
self.assertEqual(message.get("subject"), "Subject")
|
||
self.assertEqual(message.get_all("to"), ["nobody@example.com"])
|
||
self.assertTrue(message.is_multipart())
|
||
self.assertEqual(len(message.get_payload()), 2)
|
||
self.assertEqual(message.get_payload(0).get_payload(), "Content")
|
||
self.assertEqual(message.get_payload(0).get_content_type(), "text/plain")
|
||
self.assertEqual(message.get_payload(1).get_payload(), "HTML Content")
|
||
self.assertEqual(message.get_payload(1).get_content_type(), "text/html")
|
||
|
||
@override_settings(MANAGERS=[("nobody", "nobody@example.com")])
|
||
def test_html_mail_managers(self):
|
||
"""Test html_message argument to mail_managers"""
|
||
mail_managers("Subject", "Content", html_message="HTML Content")
|
||
message = self.get_the_message()
|
||
|
||
self.assertEqual(message.get("subject"), "[Django] Subject")
|
||
self.assertEqual(message.get_all("to"), ["nobody@example.com"])
|
||
self.assertTrue(message.is_multipart())
|
||
self.assertEqual(len(message.get_payload()), 2)
|
||
self.assertEqual(message.get_payload(0).get_payload(), "Content")
|
||
self.assertEqual(message.get_payload(0).get_content_type(), "text/plain")
|
||
self.assertEqual(message.get_payload(1).get_payload(), "HTML Content")
|
||
self.assertEqual(message.get_payload(1).get_content_type(), "text/html")
|
||
|
||
@override_settings(ADMINS=[("nobody", "nobody@example.com")])
|
||
def test_html_mail_admins(self):
|
||
"""Test html_message argument to mail_admins"""
|
||
mail_admins("Subject", "Content", html_message="HTML Content")
|
||
message = self.get_the_message()
|
||
|
||
self.assertEqual(message.get("subject"), "[Django] Subject")
|
||
self.assertEqual(message.get_all("to"), ["nobody@example.com"])
|
||
self.assertTrue(message.is_multipart())
|
||
self.assertEqual(len(message.get_payload()), 2)
|
||
self.assertEqual(message.get_payload(0).get_payload(), "Content")
|
||
self.assertEqual(message.get_payload(0).get_content_type(), "text/plain")
|
||
self.assertEqual(message.get_payload(1).get_payload(), "HTML Content")
|
||
self.assertEqual(message.get_payload(1).get_content_type(), "text/html")
|
||
|
||
@override_settings(
|
||
ADMINS=[("nobody", "nobody+admin@example.com")],
|
||
MANAGERS=[("nobody", "nobody+manager@example.com")],
|
||
)
|
||
def test_manager_and_admin_mail_prefix(self):
|
||
"""
|
||
String prefix + lazy translated subject = bad output
|
||
Regression for #13494
|
||
"""
|
||
mail_managers(gettext_lazy("Subject"), "Content")
|
||
message = self.get_the_message()
|
||
self.assertEqual(message.get("subject"), "[Django] Subject")
|
||
|
||
self.flush_mailbox()
|
||
mail_admins(gettext_lazy("Subject"), "Content")
|
||
message = self.get_the_message()
|
||
self.assertEqual(message.get("subject"), "[Django] Subject")
|
||
|
||
@override_settings(ADMINS=[], MANAGERS=[])
|
||
def test_empty_admins(self):
|
||
"""
|
||
mail_admins/mail_managers doesn't connect to the mail server
|
||
if there are no recipients (#9383)
|
||
"""
|
||
mail_admins("hi", "there")
|
||
self.assertEqual(self.get_mailbox_content(), [])
|
||
mail_managers("hi", "there")
|
||
self.assertEqual(self.get_mailbox_content(), [])
|
||
|
||
def test_wrong_admins_managers(self):
|
||
tests = (
|
||
"test@example.com",
|
||
("test@example.com",),
|
||
["test@example.com", "other@example.com"],
|
||
("test@example.com", "other@example.com"),
|
||
)
|
||
for setting, mail_func in (
|
||
("ADMINS", mail_admins),
|
||
("MANAGERS", mail_managers),
|
||
):
|
||
msg = "The %s setting must be a list of 2-tuples." % setting
|
||
for value in tests:
|
||
with (
|
||
self.subTest(setting=setting, value=value),
|
||
self.settings(**{setting: value}),
|
||
):
|
||
with self.assertRaisesMessage(ValueError, msg):
|
||
mail_func("subject", "content")
|
||
|
||
def test_message_cc_header(self):
|
||
"""
|
||
Regression test for #7722
|
||
"""
|
||
email = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com"],
|
||
cc=["cc@example.com"],
|
||
)
|
||
mail.get_connection().send_messages([email])
|
||
message = self.get_the_message()
|
||
self.assertMessageHasHeaders(
|
||
message,
|
||
{
|
||
("MIME-Version", "1.0"),
|
||
("Content-Type", 'text/plain; charset="utf-8"'),
|
||
("Content-Transfer-Encoding", "7bit"),
|
||
("Subject", "Subject"),
|
||
("From", "from@example.com"),
|
||
("To", "to@example.com"),
|
||
("Cc", "cc@example.com"),
|
||
},
|
||
)
|
||
self.assertIn("\nDate: ", message.as_string())
|
||
|
||
def test_idn_send(self):
|
||
"""
|
||
Regression test for #14301
|
||
"""
|
||
self.assertTrue(send_mail("Subject", "Content", "from@öäü.com", ["to@öäü.com"]))
|
||
message = self.get_the_message()
|
||
self.assertEqual(message.get("from"), "from@xn--4ca9at.com")
|
||
self.assertEqual(message.get("to"), "to@xn--4ca9at.com")
|
||
|
||
self.flush_mailbox()
|
||
m = EmailMessage(
|
||
from_email="from@öäü.com", to=["to@öäü.com"], cc=["cc@öäü.com"]
|
||
)
|
||
m.send()
|
||
message = self.get_the_message()
|
||
self.assertEqual(message.get("from"), "from@xn--4ca9at.com")
|
||
self.assertEqual(message.get("to"), "to@xn--4ca9at.com")
|
||
self.assertEqual(message.get("cc"), "cc@xn--4ca9at.com")
|
||
|
||
def test_recipient_without_domain(self):
|
||
"""
|
||
Regression test for #15042
|
||
"""
|
||
self.assertTrue(send_mail("Subject", "Content", "tester", ["django"]))
|
||
message = self.get_the_message()
|
||
self.assertEqual(message.get("from"), "tester")
|
||
self.assertEqual(message.get("to"), "django")
|
||
|
||
def test_lazy_addresses(self):
|
||
"""
|
||
Email sending should support lazy email addresses (#24416).
|
||
"""
|
||
_ = gettext_lazy
|
||
self.assertTrue(send_mail("Subject", "Content", _("tester"), [_("django")]))
|
||
message = self.get_the_message()
|
||
self.assertEqual(message.get("from"), "tester")
|
||
self.assertEqual(message.get("to"), "django")
|
||
|
||
self.flush_mailbox()
|
||
m = EmailMessage(
|
||
from_email=_("tester"),
|
||
to=[_("to1"), _("to2")],
|
||
cc=[_("cc1"), _("cc2")],
|
||
bcc=[_("bcc")],
|
||
reply_to=[_("reply")],
|
||
)
|
||
self.assertEqual(m.recipients(), ["to1", "to2", "cc1", "cc2", "bcc"])
|
||
m.send()
|
||
message = self.get_the_message()
|
||
self.assertEqual(message.get("from"), "tester")
|
||
self.assertEqual(message.get("to"), "to1, to2")
|
||
self.assertEqual(message.get("cc"), "cc1, cc2")
|
||
self.assertEqual(message.get("Reply-To"), "reply")
|
||
|
||
def test_close_connection(self):
|
||
"""
|
||
Connection can be closed (even when not explicitly opened)
|
||
"""
|
||
conn = mail.get_connection(username="", password="")
|
||
conn.close()
|
||
|
||
def test_use_as_contextmanager(self):
|
||
"""
|
||
The connection can be used as a contextmanager.
|
||
"""
|
||
opened = [False]
|
||
closed = [False]
|
||
conn = mail.get_connection(username="", password="")
|
||
|
||
def open():
|
||
opened[0] = True
|
||
|
||
conn.open = open
|
||
|
||
def close():
|
||
closed[0] = True
|
||
|
||
conn.close = close
|
||
with conn as same_conn:
|
||
self.assertTrue(opened[0])
|
||
self.assertIs(same_conn, conn)
|
||
self.assertFalse(closed[0])
|
||
self.assertTrue(closed[0])
|
||
|
||
|
||
class LocmemBackendTests(BaseEmailBackendTests, SimpleTestCase):
|
||
email_backend = "django.core.mail.backends.locmem.EmailBackend"
|
||
|
||
def get_mailbox_content(self):
|
||
# Reparse as modern messages to work with shared BaseEmailBackendTests.
|
||
# (Once EmailMessage.message() uses Python's modern email API, this
|
||
# can be changed back to `[m.message() for m in mail.outbox]`.)
|
||
return [message_from_bytes(m.message().as_bytes()) for m in mail.outbox]
|
||
|
||
def flush_mailbox(self):
|
||
mail.outbox = []
|
||
|
||
def tearDown(self):
|
||
super().tearDown()
|
||
mail.outbox = []
|
||
|
||
def test_locmem_shared_messages(self):
|
||
"""
|
||
Make sure that the locmen backend populates the outbox.
|
||
"""
|
||
connection = locmem.EmailBackend()
|
||
connection2 = locmem.EmailBackend()
|
||
email = EmailMessage(to=["to@example.com"])
|
||
connection.send_messages([email])
|
||
connection2.send_messages([email])
|
||
self.assertEqual(len(mail.outbox), 2)
|
||
|
||
def test_validate_multiline_headers(self):
|
||
# Ticket #18861 - Validate emails when using the locmem backend
|
||
with self.assertRaises(BadHeaderError):
|
||
send_mail(
|
||
"Subject\nMultiline", "Content", "from@example.com", ["to@example.com"]
|
||
)
|
||
|
||
def test_outbox_not_mutated_after_send(self):
|
||
email = EmailMessage(
|
||
subject="correct subject",
|
||
to=["to@example.com"],
|
||
)
|
||
email.send()
|
||
email.subject = "other subject"
|
||
email.to.append("other@example.com")
|
||
self.assertEqual(mail.outbox[0].subject, "correct subject")
|
||
self.assertEqual(mail.outbox[0].to, ["to@example.com"])
|
||
|
||
|
||
class FileBackendTests(BaseEmailBackendTests, SimpleTestCase):
|
||
email_backend = "django.core.mail.backends.filebased.EmailBackend"
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.tmp_dir = self.mkdtemp()
|
||
self.addCleanup(shutil.rmtree, self.tmp_dir)
|
||
_settings_override = override_settings(EMAIL_FILE_PATH=self.tmp_dir)
|
||
_settings_override.enable()
|
||
self.addCleanup(_settings_override.disable)
|
||
|
||
def mkdtemp(self):
|
||
return tempfile.mkdtemp()
|
||
|
||
def flush_mailbox(self):
|
||
for filename in os.listdir(self.tmp_dir):
|
||
os.unlink(os.path.join(self.tmp_dir, filename))
|
||
|
||
def get_mailbox_content(self):
|
||
messages = []
|
||
for filename in os.listdir(self.tmp_dir):
|
||
with open(os.path.join(self.tmp_dir, filename), "rb") as fp:
|
||
session = fp.read().split(b"\n" + (b"-" * 79) + b"\n")
|
||
messages.extend(message_from_bytes(m) for m in session if m)
|
||
return messages
|
||
|
||
def test_file_sessions(self):
|
||
"""Make sure opening a connection creates a new file"""
|
||
msg = EmailMessage(
|
||
"Subject",
|
||
"Content",
|
||
"bounce@example.com",
|
||
["to@example.com"],
|
||
headers={"From": "from@example.com"},
|
||
)
|
||
connection = mail.get_connection()
|
||
connection.send_messages([msg])
|
||
|
||
self.assertEqual(len(os.listdir(self.tmp_dir)), 1)
|
||
with open(os.path.join(self.tmp_dir, os.listdir(self.tmp_dir)[0]), "rb") as fp:
|
||
message = message_from_binary_file(fp, policy=policy.default)
|
||
self.assertEqual(message.get_content_type(), "text/plain")
|
||
self.assertEqual(message.get("subject"), "Subject")
|
||
self.assertEqual(message.get("from"), "from@example.com")
|
||
self.assertEqual(message.get("to"), "to@example.com")
|
||
|
||
connection2 = mail.get_connection()
|
||
connection2.send_messages([msg])
|
||
self.assertEqual(len(os.listdir(self.tmp_dir)), 2)
|
||
|
||
connection.send_messages([msg])
|
||
self.assertEqual(len(os.listdir(self.tmp_dir)), 2)
|
||
|
||
msg.connection = mail.get_connection()
|
||
self.assertTrue(connection.open())
|
||
msg.send()
|
||
self.assertEqual(len(os.listdir(self.tmp_dir)), 3)
|
||
msg.send()
|
||
self.assertEqual(len(os.listdir(self.tmp_dir)), 3)
|
||
|
||
connection.close()
|
||
|
||
|
||
class FileBackendPathLibTests(FileBackendTests):
|
||
def mkdtemp(self):
|
||
tmp_dir = super().mkdtemp()
|
||
return Path(tmp_dir)
|
||
|
||
|
||
class ConsoleBackendTests(BaseEmailBackendTests, SimpleTestCase):
|
||
email_backend = "django.core.mail.backends.console.EmailBackend"
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.__stdout = sys.stdout
|
||
self.stream = sys.stdout = StringIO()
|
||
|
||
def tearDown(self):
|
||
del self.stream
|
||
sys.stdout = self.__stdout
|
||
del self.__stdout
|
||
super().tearDown()
|
||
|
||
def flush_mailbox(self):
|
||
self.stream = sys.stdout = StringIO()
|
||
|
||
def get_mailbox_content(self):
|
||
messages = self.stream.getvalue().split("\n" + ("-" * 79) + "\n")
|
||
return [message_from_bytes(m.encode()) for m in messages if m]
|
||
|
||
def test_console_stream_kwarg(self):
|
||
"""
|
||
The console backend can be pointed at an arbitrary stream.
|
||
"""
|
||
s = StringIO()
|
||
connection = mail.get_connection(
|
||
"django.core.mail.backends.console.EmailBackend", stream=s
|
||
)
|
||
send_mail(
|
||
"Subject",
|
||
"Content",
|
||
"from@example.com",
|
||
["to@example.com"],
|
||
connection=connection,
|
||
)
|
||
message = s.getvalue().split("\n" + ("-" * 79) + "\n")[0].encode()
|
||
self.assertMessageHasHeaders(
|
||
message,
|
||
{
|
||
("MIME-Version", "1.0"),
|
||
("Content-Type", 'text/plain; charset="utf-8"'),
|
||
("Content-Transfer-Encoding", "7bit"),
|
||
("Subject", "Subject"),
|
||
("From", "from@example.com"),
|
||
("To", "to@example.com"),
|
||
},
|
||
)
|
||
self.assertIn(b"\nDate: ", message)
|
||
|
||
|
||
class SMTPHandler:
|
||
def __init__(self, *args, **kwargs):
|
||
self.mailbox = []
|
||
self.smtp_envelopes = []
|
||
|
||
async def handle_DATA(self, server, session, envelope):
|
||
data = envelope.content
|
||
mail_from = envelope.mail_from
|
||
|
||
message = message_from_bytes(data.rstrip())
|
||
try:
|
||
header_from = message["from"].addresses[0].addr_spec
|
||
except (KeyError, IndexError):
|
||
header_from = None
|
||
|
||
if mail_from != header_from:
|
||
return f"553 '{mail_from}' != '{header_from}'"
|
||
self.mailbox.append(message)
|
||
self.smtp_envelopes.append(
|
||
{
|
||
"mail_from": envelope.mail_from,
|
||
"rcpt_tos": envelope.rcpt_tos,
|
||
}
|
||
)
|
||
return "250 OK"
|
||
|
||
def flush_mailbox(self):
|
||
self.mailbox[:] = []
|
||
self.smtp_envelopes[:] = []
|
||
|
||
|
||
@skipUnless(HAS_AIOSMTPD, "No aiosmtpd library detected.")
|
||
class SMTPBackendTestsBase(SimpleTestCase):
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
super().setUpClass()
|
||
# Find a free port.
|
||
with socket.socket() as s:
|
||
s.bind(("127.0.0.1", 0))
|
||
port = s.getsockname()[1]
|
||
cls.smtp_handler = SMTPHandler()
|
||
cls.smtp_controller = Controller(
|
||
cls.smtp_handler,
|
||
hostname="127.0.0.1",
|
||
port=port,
|
||
)
|
||
cls._settings_override = override_settings(
|
||
EMAIL_HOST=cls.smtp_controller.hostname,
|
||
EMAIL_PORT=cls.smtp_controller.port,
|
||
)
|
||
cls._settings_override.enable()
|
||
cls.addClassCleanup(cls._settings_override.disable)
|
||
cls.smtp_controller.start()
|
||
cls.addClassCleanup(cls.stop_smtp)
|
||
|
||
@classmethod
|
||
def stop_smtp(cls):
|
||
cls.smtp_controller.stop()
|
||
|
||
|
||
@skipUnless(HAS_AIOSMTPD, "No aiosmtpd library detected.")
|
||
class SMTPBackendTests(BaseEmailBackendTests, SMTPBackendTestsBase):
|
||
email_backend = "django.core.mail.backends.smtp.EmailBackend"
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.smtp_handler.flush_mailbox()
|
||
self.addCleanup(self.smtp_handler.flush_mailbox)
|
||
|
||
def flush_mailbox(self):
|
||
self.smtp_handler.flush_mailbox()
|
||
|
||
def get_mailbox_content(self):
|
||
return self.smtp_handler.mailbox
|
||
|
||
def get_smtp_envelopes(self):
|
||
return self.smtp_handler.smtp_envelopes
|
||
|
||
@override_settings(
|
||
EMAIL_HOST_USER="not empty username",
|
||
EMAIL_HOST_PASSWORD="not empty password",
|
||
)
|
||
def test_email_authentication_use_settings(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertEqual(backend.username, "not empty username")
|
||
self.assertEqual(backend.password, "not empty password")
|
||
|
||
@override_settings(
|
||
EMAIL_HOST_USER="not empty username",
|
||
EMAIL_HOST_PASSWORD="not empty password",
|
||
)
|
||
def test_email_authentication_override_settings(self):
|
||
backend = smtp.EmailBackend(username="username", password="password")
|
||
self.assertEqual(backend.username, "username")
|
||
self.assertEqual(backend.password, "password")
|
||
|
||
@override_settings(
|
||
EMAIL_HOST_USER="not empty username",
|
||
EMAIL_HOST_PASSWORD="not empty password",
|
||
)
|
||
def test_email_disabled_authentication(self):
|
||
backend = smtp.EmailBackend(username="", password="")
|
||
self.assertEqual(backend.username, "")
|
||
self.assertEqual(backend.password, "")
|
||
|
||
def test_auth_attempted(self):
|
||
"""
|
||
Opening the backend with non empty username/password tries
|
||
to authenticate against the SMTP server.
|
||
"""
|
||
backend = smtp.EmailBackend(
|
||
username="not empty username", password="not empty password"
|
||
)
|
||
with self.assertRaisesMessage(
|
||
SMTPException, "SMTP AUTH extension not supported by server."
|
||
):
|
||
with backend:
|
||
pass
|
||
|
||
def test_server_open(self):
|
||
"""
|
||
open() returns whether it opened a connection.
|
||
"""
|
||
backend = smtp.EmailBackend(username="", password="")
|
||
self.assertIsNone(backend.connection)
|
||
opened = backend.open()
|
||
backend.close()
|
||
self.assertIs(opened, True)
|
||
|
||
def test_reopen_connection(self):
|
||
backend = smtp.EmailBackend()
|
||
# Simulate an already open connection.
|
||
backend.connection = mock.Mock(spec=object())
|
||
self.assertIs(backend.open(), False)
|
||
|
||
@override_settings(EMAIL_USE_TLS=True)
|
||
def test_email_tls_use_settings(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertTrue(backend.use_tls)
|
||
|
||
@override_settings(EMAIL_USE_TLS=True)
|
||
def test_email_tls_override_settings(self):
|
||
backend = smtp.EmailBackend(use_tls=False)
|
||
self.assertFalse(backend.use_tls)
|
||
|
||
def test_email_tls_default_disabled(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertFalse(backend.use_tls)
|
||
|
||
def test_ssl_tls_mutually_exclusive(self):
|
||
msg = (
|
||
"EMAIL_USE_TLS/EMAIL_USE_SSL are mutually exclusive, so only set "
|
||
"one of those settings to True."
|
||
)
|
||
with self.assertRaisesMessage(ValueError, msg):
|
||
smtp.EmailBackend(use_ssl=True, use_tls=True)
|
||
|
||
@override_settings(EMAIL_USE_SSL=True)
|
||
def test_email_ssl_use_settings(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertTrue(backend.use_ssl)
|
||
|
||
@override_settings(EMAIL_USE_SSL=True)
|
||
def test_email_ssl_override_settings(self):
|
||
backend = smtp.EmailBackend(use_ssl=False)
|
||
self.assertFalse(backend.use_ssl)
|
||
|
||
def test_email_ssl_default_disabled(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertFalse(backend.use_ssl)
|
||
|
||
@override_settings(EMAIL_SSL_CERTFILE="foo")
|
||
def test_email_ssl_certfile_use_settings(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertEqual(backend.ssl_certfile, "foo")
|
||
|
||
@override_settings(EMAIL_SSL_CERTFILE="foo")
|
||
def test_email_ssl_certfile_override_settings(self):
|
||
backend = smtp.EmailBackend(ssl_certfile="bar")
|
||
self.assertEqual(backend.ssl_certfile, "bar")
|
||
|
||
def test_email_ssl_certfile_default_disabled(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertIsNone(backend.ssl_certfile)
|
||
|
||
@override_settings(EMAIL_SSL_KEYFILE="foo")
|
||
def test_email_ssl_keyfile_use_settings(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertEqual(backend.ssl_keyfile, "foo")
|
||
|
||
@override_settings(EMAIL_SSL_KEYFILE="foo")
|
||
def test_email_ssl_keyfile_override_settings(self):
|
||
backend = smtp.EmailBackend(ssl_keyfile="bar")
|
||
self.assertEqual(backend.ssl_keyfile, "bar")
|
||
|
||
def test_email_ssl_keyfile_default_disabled(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertIsNone(backend.ssl_keyfile)
|
||
|
||
@override_settings(EMAIL_USE_TLS=True)
|
||
def test_email_tls_attempts_starttls(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertTrue(backend.use_tls)
|
||
with self.assertRaisesMessage(
|
||
SMTPException, "STARTTLS extension not supported by server."
|
||
):
|
||
with backend:
|
||
pass
|
||
|
||
@override_settings(EMAIL_USE_SSL=True)
|
||
def test_email_ssl_attempts_ssl_connection(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertTrue(backend.use_ssl)
|
||
with self.assertRaises(SSLError):
|
||
with backend:
|
||
pass
|
||
|
||
def test_connection_timeout_default(self):
|
||
"""The connection's timeout value is None by default."""
|
||
connection = mail.get_connection("django.core.mail.backends.smtp.EmailBackend")
|
||
self.assertIsNone(connection.timeout)
|
||
|
||
def test_connection_timeout_custom(self):
|
||
"""The timeout parameter can be customized."""
|
||
|
||
class MyEmailBackend(smtp.EmailBackend):
|
||
def __init__(self, *args, **kwargs):
|
||
kwargs.setdefault("timeout", 42)
|
||
super().__init__(*args, **kwargs)
|
||
|
||
myemailbackend = MyEmailBackend()
|
||
myemailbackend.open()
|
||
self.assertEqual(myemailbackend.timeout, 42)
|
||
self.assertEqual(myemailbackend.connection.timeout, 42)
|
||
myemailbackend.close()
|
||
|
||
@override_settings(EMAIL_TIMEOUT=10)
|
||
def test_email_timeout_override_settings(self):
|
||
backend = smtp.EmailBackend()
|
||
self.assertEqual(backend.timeout, 10)
|
||
|
||
def test_email_msg_uses_crlf(self):
|
||
"""#23063 -- RFC-compliant messages are sent over SMTP."""
|
||
send = SMTP.send
|
||
try:
|
||
smtp_messages = []
|
||
|
||
def mock_send(self, s):
|
||
smtp_messages.append(s)
|
||
return send(self, s)
|
||
|
||
SMTP.send = mock_send
|
||
|
||
email = EmailMessage(
|
||
"Subject", "Content", "from@example.com", ["to@example.com"]
|
||
)
|
||
mail.get_connection().send_messages([email])
|
||
|
||
# Find the actual message
|
||
msg = None
|
||
for i, m in enumerate(smtp_messages):
|
||
if m[:4] == "data":
|
||
msg = smtp_messages[i + 1]
|
||
break
|
||
|
||
self.assertTrue(msg)
|
||
|
||
msg = msg.decode()
|
||
# The message only contains CRLF and not combinations of CRLF, LF, and CR.
|
||
msg = msg.replace("\r\n", "")
|
||
self.assertNotIn("\r", msg)
|
||
self.assertNotIn("\n", msg)
|
||
|
||
finally:
|
||
SMTP.send = send
|
||
|
||
def test_send_messages_after_open_failed(self):
|
||
"""
|
||
send_messages() shouldn't try to send messages if open() raises an
|
||
exception after initializing the connection.
|
||
"""
|
||
backend = smtp.EmailBackend()
|
||
# Simulate connection initialization success and a subsequent
|
||
# connection exception.
|
||
backend.connection = mock.Mock(spec=object())
|
||
backend.open = lambda: None
|
||
email = EmailMessage(to=["to@example.com"])
|
||
self.assertEqual(backend.send_messages([email]), 0)
|
||
|
||
def test_send_messages_empty_list(self):
|
||
backend = smtp.EmailBackend()
|
||
backend.connection = mock.Mock(spec=object())
|
||
self.assertEqual(backend.send_messages([]), 0)
|
||
|
||
def test_send_messages_zero_sent(self):
|
||
"""A message isn't sent if it doesn't have any recipients."""
|
||
backend = smtp.EmailBackend()
|
||
backend.connection = mock.Mock(spec=object())
|
||
email = EmailMessage("Subject", "Content", "from@example.com", to=[])
|
||
sent = backend.send_messages([email])
|
||
self.assertEqual(sent, 0)
|
||
|
||
def test_avoids_sending_to_invalid_addresses(self):
|
||
"""
|
||
Verify invalid addresses can't sneak into SMTP commands through
|
||
EmailMessage.all_recipients() (which is distinct from message header fields).
|
||
"""
|
||
backend = smtp.EmailBackend()
|
||
backend.connection = mock.Mock()
|
||
for email_address in (
|
||
# Invalid address with two @ signs.
|
||
"to@other.com@example.com",
|
||
# Invalid address without the quotes.
|
||
"to@other.com <to@example.com>",
|
||
# Other invalid addresses.
|
||
"@",
|
||
"to@",
|
||
"@example.com",
|
||
# CR/NL in addr-spec. (SMTP strips display-name.)
|
||
'"evil@example.com\r\nto"@example.com',
|
||
"to\nevil@example.com",
|
||
):
|
||
with self.subTest(email_address=email_address):
|
||
# Use bcc (which is only processed by SMTP backend) to ensure
|
||
# error is coming from SMTP backend, not EmailMessage.message().
|
||
email = EmailMessage(bcc=[email_address])
|
||
with self.assertRaisesMessage(ValueError, "Invalid address"):
|
||
backend.send_messages([email])
|
||
|
||
def test_encodes_idna_in_smtp_commands(self):
|
||
"""
|
||
SMTP backend must encode non-ASCII domains for the SMTP envelope
|
||
(which can be distinct from the email headers).
|
||
"""
|
||
email = EmailMessage(
|
||
from_email="lists@discussão.example.org",
|
||
to=["To Example <to@漢字.example.com>"],
|
||
bcc=["monitor@discussão.example.org"],
|
||
headers={
|
||
"From": "Gestor de listas <lists@discussão.example.org>",
|
||
"To": "Discussão Django <django@discussão.example.org>",
|
||
},
|
||
)
|
||
backend = smtp.EmailBackend()
|
||
backend.send_messages([email])
|
||
envelope = self.get_smtp_envelopes()[0]
|
||
self.assertEqual(envelope["mail_from"], "lists@xn--discusso-xza.example.org")
|
||
self.assertEqual(
|
||
envelope["rcpt_tos"],
|
||
["to@xn--p8s937b.example.com", "monitor@xn--discusso-xza.example.org"],
|
||
)
|
||
|
||
def test_does_not_reencode_idna(self):
|
||
"""
|
||
SMTP backend should not downgrade IDNA 2008 to IDNA 2003.
|
||
|
||
Django does not currently handle IDNA 2008 encoding, but should retain
|
||
it for addresses that have been pre-encoded.
|
||
"""
|
||
# Test all four EmailMessage attrs accessed by the SMTP email backend.
|
||
# These are IDNA 2008 encoded domains that would be different
|
||
# in IDNA 2003, from https://www.unicode.org/reports/tr46/#Deviations.
|
||
email = EmailMessage(
|
||
from_email='"βόλος" <from@xn--fa-hia.example.com>',
|
||
to=['"faß" <to@xn--10cl1a0b660p.example.com>'],
|
||
cc=['"ශ්රී" <cc@xn--nxasmm1c.example.com>'],
|
||
bcc=['"نامهای." <bcc@xn--mgba3gch31f060k.example.com>'],
|
||
)
|
||
backend = smtp.EmailBackend()
|
||
backend.send_messages([email])
|
||
envelope = self.get_smtp_envelopes()[0]
|
||
self.assertEqual(envelope["mail_from"], "from@xn--fa-hia.example.com")
|
||
self.assertEqual(
|
||
envelope["rcpt_tos"],
|
||
[
|
||
"to@xn--10cl1a0b660p.example.com",
|
||
"cc@xn--nxasmm1c.example.com",
|
||
"bcc@xn--mgba3gch31f060k.example.com",
|
||
],
|
||
)
|
||
|
||
|
||
@skipUnless(HAS_AIOSMTPD, "No aiosmtpd library detected.")
|
||
class SMTPBackendStoppedServerTests(SMTPBackendTestsBase):
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
super().setUpClass()
|
||
cls.backend = smtp.EmailBackend(username="", password="")
|
||
cls.smtp_controller.stop()
|
||
|
||
@classmethod
|
||
def stop_smtp(cls):
|
||
# SMTP controller is stopped in setUpClass().
|
||
pass
|
||
|
||
def test_server_stopped(self):
|
||
"""
|
||
Closing the backend while the SMTP server is stopped doesn't raise an
|
||
exception.
|
||
"""
|
||
self.backend.close()
|
||
|
||
def test_fail_silently_on_connection_error(self):
|
||
"""
|
||
A socket connection error is silenced with fail_silently=True.
|
||
"""
|
||
with self.assertRaises(ConnectionError):
|
||
self.backend.open()
|
||
self.backend.fail_silently = True
|
||
self.backend.open()
|