mirror of
https://github.com/django/django.git
synced 2024-12-24 18:16:19 +00:00
afd040d4d3
assert_ -> assertTrue assertEquals -> assertEqual failUnless -> assertTrue For full details, see http://www.voidspace.org.uk/python/articles/unittest2.shtml#deprecations git-svn-id: http://code.djangoproject.com/svn/django/trunk@15728 bcc190cf-cafb-0310-a4f2-bffc1f526a37
637 lines
29 KiB
Python
637 lines
29 KiB
Python
# coding: utf-8
|
|
import asyncore
|
|
import email
|
|
import os
|
|
import shutil
|
|
import smtpd
|
|
import sys
|
|
from StringIO import StringIO
|
|
import tempfile
|
|
import threading
|
|
|
|
from django.conf import settings
|
|
from django.core import mail
|
|
from django.core.mail import (EmailMessage, mail_admins, mail_managers,
|
|
EmailMultiAlternatives, send_mail, send_mass_mail)
|
|
from django.core.mail.backends import console, dummy, locmem, filebased, smtp
|
|
from django.core.mail.message import BadHeaderError
|
|
from django.test import TestCase
|
|
from django.utils.translation import ugettext_lazy
|
|
from django.utils.functional import wraps
|
|
|
|
|
|
def alter_django_settings(**kwargs):
|
|
oldvalues = {}
|
|
nonexistant = []
|
|
for setting, newvalue in kwargs.iteritems():
|
|
try:
|
|
oldvalues[setting] = getattr(settings, setting)
|
|
except AttributeError:
|
|
nonexistant.append(setting)
|
|
setattr(settings, setting, newvalue)
|
|
return oldvalues, nonexistant
|
|
|
|
|
|
def restore_django_settings(state):
|
|
oldvalues, nonexistant = state
|
|
for setting, oldvalue in oldvalues.iteritems():
|
|
setattr(settings, setting, oldvalue)
|
|
for setting in nonexistant:
|
|
delattr(settings, setting)
|
|
|
|
|
|
def with_django_settings(**kwargs):
|
|
def decorator(test):
|
|
@wraps(test)
|
|
def decorated_test(self):
|
|
state = alter_django_settings(**kwargs)
|
|
try:
|
|
return test(self)
|
|
finally:
|
|
restore_django_settings(state)
|
|
return decorated_test
|
|
return decorator
|
|
|
|
|
|
class MailTests(TestCase):
|
|
"""
|
|
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'].encode(), 'Subject')
|
|
self.assertEqual(message.get_payload(), 'Content')
|
|
self.assertEqual(message['From'], 'from@example.com')
|
|
self.assertEqual(message['To'], 'to@example.com')
|
|
|
|
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'].encode(), '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_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_header_injection(self):
|
|
email = EmailMessage('Subject\nInjection Test', 'Content', 'from@example.com', ['to@example.com'])
|
|
self.assertRaises(BadHeaderError, email.message)
|
|
email = EmailMessage(ugettext_lazy('Subject\nInjection Test'), 'Content', 'from@example.com', ['to@example.com'])
|
|
self.assertRaises(BadHeaderError, email.message)
|
|
|
|
def test_space_continuation(self):
|
|
"""
|
|
Test for space continuation character in long (ascii) subject headers (#7747)
|
|
"""
|
|
email = EmailMessage('Long subject lines that get wrapped should use a space continuation character to get expected behaviour in Outlook and Thunderbird', 'Content', 'from@example.com', ['to@example.com'])
|
|
message = email.message()
|
|
self.assertEqual(message['Subject'], 'Long subject lines that get wrapped should use a space continuation\n character to get expected behaviour 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('subject', 'content', 'from@example.com', ['to@example.com'], headers=headers)
|
|
self.assertEqual(email.message().as_string(), 'Content-Type: text/plain; charset="utf-8"\nMIME-Version: 1.0\nContent-Transfer-Encoding: quoted-printable\nSubject: subject\nFrom: from@example.com\nTo: to@example.com\ndate: Fri, 09 Nov 2001 01:08:47 -0000\nMessage-ID: foo\n\ncontent')
|
|
|
|
def test_from_header(self):
|
|
"""
|
|
Make sure we can manually set the From header (#9214)
|
|
"""
|
|
email = EmailMessage('Subject', 'Content', 'bounce@example.com', ['to@example.com'], headers={'From': 'from@example.com'})
|
|
message = email.message()
|
|
self.assertEqual(message['From'], 'from@example.com')
|
|
|
|
def test_multiple_message_call(self):
|
|
"""
|
|
Regression for #13259 - Make sure that headers are not changed when
|
|
calling EmailMessage.message()
|
|
"""
|
|
email = EmailMessage('Subject', 'Content', 'bounce@example.com', ['to@example.com'], headers={'From': 'from@example.com'})
|
|
message = email.message()
|
|
self.assertEqual(message['From'], 'from@example.com')
|
|
message = email.message()
|
|
self.assertEqual(message['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('Subject', 'Content', 'from@example.com', ['"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('Subject', 'Content', 'from@example.com', ['"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(u"Gżegżółka", "Content", "from@example.com", ["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_safe_mime_multipart(self):
|
|
"""
|
|
Make sure headers can be set with a different encoding than utf-8 in
|
|
SafeMIMEMultipart as well
|
|
"""
|
|
headers = {"Date": "Fri, 09 Nov 2001 01:08:47 -0000", "Message-ID": "foo"}
|
|
subject, from_email, to = 'hello', 'from@example.com', '"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'].encode(), u'=?iso-8859-1?q?Message_from_Firstname_S=FCrname?=')
|
|
|
|
def test_encoding(self):
|
|
"""
|
|
Regression for #12791 - Encode body correctly with other encodings
|
|
than utf-8
|
|
"""
|
|
email = EmailMessage('Subject', 'Firstname Sürname is a great guy.', 'from@example.com', ['other@example.com'])
|
|
email.encoding = 'iso-8859-1'
|
|
message = email.message()
|
|
self.assertTrue(message.as_string().startswith('Content-Type: text/plain; charset="iso-8859-1"\nMIME-Version: 1.0\nContent-Transfer-Encoding: quoted-printable\nSubject: Subject\nFrom: from@example.com\nTo: other@example.com'))
|
|
self.assertEqual(message.get_payload(), 'Firstname S=FCrname is a great guy.')
|
|
|
|
# Make sure MIME attachments also works 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>'
|
|
msg = EmailMultiAlternatives('Subject', text_content, 'from@example.com', ['to@example.com'])
|
|
msg.encoding = 'iso-8859-1'
|
|
msg.attach_alternative(html_content, "text/html")
|
|
self.assertEqual(msg.message().get_payload(0).as_string(), 'Content-Type: text/plain; charset="iso-8859-1"\nMIME-Version: 1.0\nContent-Transfer-Encoding: quoted-printable\n\nFirstname S=FCrname is a great guy.')
|
|
self.assertEqual(msg.message().get_payload(1).as_string(), 'Content-Type: text/html; charset="iso-8859-1"\nMIME-Version: 1.0\nContent-Transfer-Encoding: quoted-printable\n\n<p>Firstname S=FCrname is a <strong>great</strong> guy.</p>')
|
|
|
|
def test_attachments(self):
|
|
"""Regression test for #9367"""
|
|
headers = {"Date": "Fri, 09 Nov 2001 01:08:47 -0000", "Message-ID": "foo"}
|
|
subject, from_email, to = 'hello', 'from@example.com', 'to@example.com'
|
|
text_content = 'This is an important message.'
|
|
html_content = '<p>This is an <strong>important</strong> message.</p>'
|
|
msg = EmailMultiAlternatives(subject, text_content, from_email, [to], headers=headers)
|
|
msg.attach_alternative(html_content, "text/html")
|
|
msg.attach("an attachment.pdf", "%PDF-1.4.%...", mimetype="application/pdf")
|
|
msg_str = msg.message().as_string()
|
|
message = email.message_from_string(msg_str)
|
|
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_dummy_backend(self):
|
|
"""
|
|
Make sure that dummy backends returns correct number of sent messages
|
|
"""
|
|
connection = dummy.EmailBackend()
|
|
email = EmailMessage('Subject', 'Content', 'bounce@example.com', ['to@example.com'], headers={'From': 'from@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('regressiontests.mail.custombackend.EmailBackend')
|
|
self.assertTrue(hasattr(conn, 'test_outbox'))
|
|
email = EmailMessage('Subject', 'Content', 'bounce@example.com', ['to@example.com'], headers={'From': 'from@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.assertTrue(isinstance(mail.get_connection('django.core.mail.backends.smtp.EmailBackend'), smtp.EmailBackend))
|
|
self.assertTrue(isinstance(mail.get_connection('django.core.mail.backends.locmem.EmailBackend'), locmem.EmailBackend))
|
|
self.assertTrue(isinstance(mail.get_connection('django.core.mail.backends.dummy.EmailBackend'), dummy.EmailBackend))
|
|
self.assertTrue(isinstance(mail.get_connection('django.core.mail.backends.console.EmailBackend'), console.EmailBackend))
|
|
tmp_dir = tempfile.mkdtemp()
|
|
try:
|
|
self.assertTrue(isinstance(mail.get_connection('django.core.mail.backends.filebased.EmailBackend', file_path=tmp_dir), filebased.EmailBackend))
|
|
finally:
|
|
shutil.rmtree(tmp_dir)
|
|
self.assertTrue(isinstance(mail.get_connection(), locmem.EmailBackend))
|
|
|
|
@with_django_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('regressiontests.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('regressiontests.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('regressiontests.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('regressiontests.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('Subject', 'From the future', 'bounce@example.com', ['to@example.com'], headers={'From': 'from@example.com'})
|
|
self.assertFalse('>From the future' in email.message().as_string())
|
|
|
|
|
|
class BaseEmailBackendTests(object):
|
|
email_backend = None
|
|
|
|
def setUp(self):
|
|
self.__settings_state = alter_django_settings(EMAIL_BACKEND=self.email_backend)
|
|
|
|
def tearDown(self):
|
|
restore_django_settings(self.__settings_state)
|
|
|
|
def assertStartsWith(self, first, second):
|
|
if not first.startswith(second):
|
|
self.longMessage = True
|
|
self.assertEqual(first[:len(second)], second, "First string doesn't start with the second.")
|
|
|
|
def get_mailbox_content(self):
|
|
raise NotImplementedError
|
|
|
|
def flush_mailbox(self):
|
|
raise NotImplementedError
|
|
|
|
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_many(self):
|
|
email1 = EmailMessage('Subject', 'Content1', 'from@example.com', ['to@example.com'])
|
|
email2 = EmailMessage('Subject', 'Content2', 'from@example.com', ['to@example.com'])
|
|
num_sent = mail.get_connection().send_messages([email1, email2])
|
|
self.assertEqual(num_sent, 2)
|
|
messages = self.get_mailbox_content()
|
|
self.assertEqual(len(messages), 2)
|
|
self.assertEqual(messages[0].get_payload(), "Content1")
|
|
self.assertEqual(messages[1].get_payload(), "Content2")
|
|
|
|
def test_send_verbose_name(self):
|
|
email = EmailMessage("Subject", "Content", '"Firstname Sürname" <from@example.com>',
|
|
["to@example.com"])
|
|
email.send()
|
|
message = self.get_the_message()
|
|
self.assertEqual(message["subject"], "Subject")
|
|
self.assertEqual(message.get_payload(), "Content")
|
|
self.assertEqual(message["from"], "=?utf-8?q?Firstname_S=C3=BCrname?= <from@example.com>")
|
|
|
|
@with_django_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')
|
|
|
|
@with_django_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')
|
|
|
|
@with_django_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(ugettext_lazy('Subject'), 'Content')
|
|
message = self.get_the_message()
|
|
self.assertEqual(message.get('subject'), '[Django] Subject')
|
|
|
|
self.flush_mailbox()
|
|
mail_admins(ugettext_lazy('Subject'), 'Content')
|
|
message = self.get_the_message()
|
|
self.assertEqual(message.get('subject'), '[Django] Subject')
|
|
|
|
@with_django_settings(ADMINS=(), MANAGERS=())
|
|
def test_empty_admins(self):
|
|
"""
|
|
Test that 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_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.assertStartsWith(message.as_string(), 'Content-Type: text/plain; charset="utf-8"\nMIME-Version: 1.0\nContent-Transfer-Encoding: quoted-printable\nSubject: Subject\nFrom: from@example.com\nTo: to@example.com\nCc: cc@example.com\nDate: ')
|
|
|
|
def test_idn_send(self):
|
|
"""
|
|
Regression test for #14301
|
|
"""
|
|
self.assertTrue(send_mail('Subject', 'Content', 'from@öäü.com', [u'to@öäü.com']))
|
|
message = self.get_the_message()
|
|
self.assertEqual(message.get('subject'), 'Subject')
|
|
self.assertEqual(message.get('from'), 'from@xn--4ca9at.com')
|
|
self.assertEqual(message.get('to'), 'to@xn--4ca9at.com')
|
|
|
|
self.flush_mailbox()
|
|
m = EmailMessage('Subject', 'Content', 'from@öäü.com',
|
|
[u'to@öäü.com'], cc=[u'cc@öäü.com'])
|
|
m.send()
|
|
message = self.get_the_message()
|
|
self.assertEqual(message.get('subject'), 'Subject')
|
|
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('subject'), 'Subject')
|
|
self.assertEqual(message.get('from'), "tester")
|
|
self.assertEqual(message.get('to'), "django")
|
|
|
|
|
|
class LocmemBackendTests(BaseEmailBackendTests, TestCase):
|
|
email_backend = 'django.core.mail.backends.locmem.EmailBackend'
|
|
|
|
def get_mailbox_content(self):
|
|
return [m.message() for m in mail.outbox]
|
|
|
|
def flush_mailbox(self):
|
|
mail.outbox = []
|
|
|
|
def tearDown(self):
|
|
super(LocmemBackendTests, self).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('Subject', 'Content', 'bounce@example.com', ['to@example.com'], headers={'From': 'from@example.com'})
|
|
connection.send_messages([email])
|
|
connection2.send_messages([email])
|
|
self.assertEqual(len(mail.outbox), 2)
|
|
|
|
|
|
class FileBackendTests(BaseEmailBackendTests, TestCase):
|
|
email_backend = 'django.core.mail.backends.filebased.EmailBackend'
|
|
|
|
def setUp(self):
|
|
super(FileBackendTests, self).setUp()
|
|
self.tmp_dir = tempfile.mkdtemp()
|
|
self.__settings_state = alter_django_settings(EMAIL_FILE_PATH=self.tmp_dir)
|
|
|
|
def tearDown(self):
|
|
restore_django_settings(self.__settings_state)
|
|
shutil.rmtree(self.tmp_dir)
|
|
super(FileBackendTests, self).tearDown()
|
|
|
|
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):
|
|
session = open(os.path.join(self.tmp_dir, filename)).read().split('\n' + ('-' * 79) + '\n')
|
|
messages.extend(email.message_from_string(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)
|
|
message = email.message_from_file(open(os.path.join(self.tmp_dir, os.listdir(self.tmp_dir)[0])))
|
|
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)
|
|
|
|
|
|
class ConsoleBackendTests(BaseEmailBackendTests, TestCase):
|
|
email_backend = 'django.core.mail.backends.console.EmailBackend'
|
|
|
|
def setUp(self):
|
|
super(ConsoleBackendTests, self).setUp()
|
|
self.__stdout = sys.stdout
|
|
self.stream = sys.stdout = StringIO()
|
|
|
|
def tearDown(self):
|
|
del self.stream
|
|
sys.stdout = self.__stdout
|
|
del self.__stdout
|
|
super(ConsoleBackendTests, self).tearDown()
|
|
|
|
def flush_mailbox(self):
|
|
self.stream = sys.stdout = StringIO()
|
|
|
|
def get_mailbox_content(self):
|
|
messages = self.stream.getvalue().split('\n' + ('-' * 79) + '\n')
|
|
return [email.message_from_string(m) for m in messages if m]
|
|
|
|
def test_console_stream_kwarg(self):
|
|
"""
|
|
Test that 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)
|
|
self.assertTrue(s.getvalue().startswith('Content-Type: text/plain; charset="utf-8"\nMIME-Version: 1.0\nContent-Transfer-Encoding: quoted-printable\nSubject: Subject\nFrom: from@example.com\nTo: to@example.com\nDate: '))
|
|
|
|
|
|
class FakeSMTPServer(smtpd.SMTPServer, threading.Thread):
|
|
"""
|
|
Asyncore SMTP server wrapped into a thread. Based on DummyFTPServer from:
|
|
http://svn.python.org/view/python/branches/py3k/Lib/test/test_ftplib.py?revision=86061&view=markup
|
|
"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
threading.Thread.__init__(self)
|
|
smtpd.SMTPServer.__init__(self, *args, **kwargs)
|
|
self._sink = []
|
|
self.active = False
|
|
self.active_lock = threading.Lock()
|
|
self.sink_lock = threading.Lock()
|
|
|
|
def process_message(self, peer, mailfrom, rcpttos, data):
|
|
m = email.message_from_string(data)
|
|
maddr = email.Utils.parseaddr(m.get('from'))[1]
|
|
if mailfrom != maddr:
|
|
return "553 '%s' != '%s'" % (mailfrom, maddr)
|
|
self.sink_lock.acquire()
|
|
self._sink.append(m)
|
|
self.sink_lock.release()
|
|
|
|
def get_sink(self):
|
|
self.sink_lock.acquire()
|
|
try:
|
|
return self._sink[:]
|
|
finally:
|
|
self.sink_lock.release()
|
|
|
|
def flush_sink(self):
|
|
self.sink_lock.acquire()
|
|
self._sink[:] = []
|
|
self.sink_lock.release()
|
|
|
|
def start(self):
|
|
assert not self.active
|
|
self.__flag = threading.Event()
|
|
threading.Thread.start(self)
|
|
self.__flag.wait()
|
|
|
|
def run(self):
|
|
self.active = True
|
|
self.__flag.set()
|
|
while self.active and asyncore.socket_map:
|
|
self.active_lock.acquire()
|
|
asyncore.loop(timeout=0.1, count=1)
|
|
self.active_lock.release()
|
|
asyncore.close_all()
|
|
|
|
def stop(self):
|
|
assert self.active
|
|
self.active = False
|
|
self.join()
|
|
|
|
|
|
class SMTPBackendTests(BaseEmailBackendTests, TestCase):
|
|
email_backend = 'django.core.mail.backends.smtp.EmailBackend'
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.server = FakeSMTPServer(('127.0.0.1', 0), None)
|
|
cls.settings = alter_django_settings(
|
|
EMAIL_HOST="127.0.0.1",
|
|
EMAIL_PORT=cls.server.socket.getsockname()[1])
|
|
cls.server.start()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.stop()
|
|
|
|
def setUp(self):
|
|
super(SMTPBackendTests, self).setUp()
|
|
self.server.flush_sink()
|
|
|
|
def tearDown(self):
|
|
self.server.flush_sink()
|
|
super(SMTPBackendTests, self).tearDown()
|
|
|
|
def flush_mailbox(self):
|
|
self.server.flush_sink()
|
|
|
|
def get_mailbox_content(self):
|
|
return self.server.get_sink()
|