1
0
Fork 1
mirror of https://github.com/coddingtonbear/django-mailbox.git synced 2026-07-10 06:48:19 +02:00

adds testable repo with added features

This commit is contained in:
Joe 2021-02-03 23:05:48 -05:00
parent 62805f324e
commit c7997ebd9b
58 changed files with 2421 additions and 1541 deletions

View file

@ -17,39 +17,37 @@ def get_email_as_text(name):
with open(
os.path.join(
os.path.dirname(__file__),
'messages',
"messages",
name,
),
'rb'
"rb",
) as f:
return f.read()
class EmailMessageTestCase(TestCase):
ALLOWED_EXTRA_HEADERS = [
'MIME-Version',
'Content-Transfer-Encoding',
"MIME-Version",
"Content-Transfer-Encoding",
]
def setUp(self):
dm_settings = utils.get_settings()
self._ALLOWED_MIMETYPES = dm_settings['allowed_mimetypes']
self._STRIP_UNALLOWED_MIMETYPES = (
dm_settings['strip_unallowed_mimetypes']
)
self._TEXT_STORED_MIMETYPES = dm_settings['text_stored_mimetypes']
self._ALLOWED_MIMETYPES = dm_settings["allowed_mimetypes"]
self._STRIP_UNALLOWED_MIMETYPES = dm_settings["strip_unallowed_mimetypes"]
self._TEXT_STORED_MIMETYPES = dm_settings["text_stored_mimetypes"]
self.mailbox = Mailbox.objects.create(from_email='from@example.com')
self.mailbox = Mailbox.objects.create(from_email="from@example.com")
self.test_account = os.environ.get('EMAIL_ACCOUNT')
self.test_password = os.environ.get('EMAIL_PASSWORD')
self.test_smtp_server = os.environ.get('EMAIL_SMTP_SERVER')
self.test_from_email = 'nobody@nowhere.com'
self.test_account = os.environ.get("EMAIL_ACCOUNT")
self.test_password = os.environ.get("EMAIL_PASSWORD")
self.test_smtp_server = os.environ.get("EMAIL_SMTP_SERVER")
self.test_from_email = "nobody@nowhere.com"
self.maximum_wait_seconds = 60 * 5
settings.EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
settings.EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend"
settings.EMAIL_HOST = self.test_smtp_server
settings.EMAIL_PORT = 587
settings.EMAIL_HOST_USER = self.test_account
@ -80,10 +78,10 @@ class EmailMessageTestCase(TestCase):
with open(
os.path.join(
os.path.dirname(__file__),
'messages',
"messages",
name,
),
'rb'
"rb",
) as f:
return f.read()
@ -92,26 +90,26 @@ class EmailMessageTestCase(TestCase):
return email.message_from_bytes(copy)
def _headers_identical(self, left, right, header=None):
""" Check if headers are (close enough to) identical.
"""Check if headers are (close enough to) identical.
* This is particularly tricky because Python 2.6, Python 2.7 and
Python 3 each handle header strings slightly differently. This
should mash away all of the differences, though.
* This also has a small loophole in that when re-writing e-mail
payload encodings, we re-build the Content-Type header, so if the
header was originally unquoted, it will be quoted when rehydrating
the e-mail message.
* This is particularly tricky because Python 2.6, Python 2.7 and
Python 3 each handle header strings slightly differently. This
should mash away all of the differences, though.
* This also has a small loophole in that when re-writing e-mail
payload encodings, we re-build the Content-Type header, so if the
header was originally unquoted, it will be quoted when rehydrating
the e-mail message.
"""
if header.lower() == 'content-type':
if header.lower() == "content-type":
# Special case; given that we re-write the header, we'll be quoting
# the new content type; we need to make sure that doesn't cause
# this comparison to fail. Also, the case of the encoding could
# be changed, etc. etc. etc.
left = left.replace('"', '').upper()
right = right.replace('"', '').upper()
left = left.replace('\n\t', ' ').replace('\n ', ' ')
right = right.replace('\n\t', ' ').replace('\n ', ' ')
left = left.replace('"', "").upper()
right = right.replace('"', "").upper()
left = left.replace("\n\t", " ").replace("\n ", " ")
right = right.replace("\n\t", " ").replace("\n ", " ")
if right != left:
return False
return True
@ -152,10 +150,7 @@ class EmailMessageTestCase(TestCase):
if len(left_payloads) != len(right_payloads):
self._raise_mismatched(left, right)
for n in range(len(left_payloads)):
self.compare_email_objects(
left_payloads[n],
right_payloads[n]
)
self.compare_email_objects(left_payloads[n], right_payloads[n])
else:
if left.get_payload() is None or right.get_payload() is None:
if left.get_payload() is None:
@ -170,8 +165,7 @@ class EmailMessageTestCase(TestCase):
def _raise_mismatched(self, left, right):
raise AssertionError(
"Message payloads do not match:\n{}\n{}".format(
left.as_string(),
right.as_string()
left.as_string(), right.as_string()
)
)

View file

@ -1,12 +1,12 @@
DATABASES = {
'default': {
'NAME': 'db.sqlite3',
'ENGINE': 'django.db.backends.sqlite3',
"default": {
"NAME": "db.sqlite3",
"ENGINE": "django.db.backends.sqlite3",
},
}
INSTALLED_APPS = [
'django.contrib.auth',
'django.contrib.contenttypes',
'django_mailbox',
"django.contrib.auth",
"django.contrib.contenttypes",
"django_mailbox",
]
SECRET_KEY = 'beepboop'
SECRET_KEY = "beepboop"

View file

@ -8,16 +8,14 @@ from django_mailbox.models import Mailbox
from django_mailbox.tests.base import EmailMessageTestCase
__all__ = ['TestImap']
__all__ = ["TestImap"]
class TestImap(EmailMessageTestCase):
def setUp(self):
super().setUp()
self.test_imap_server = (
os.environ.get('EMAIL_IMAP_SERVER')
)
self.test_imap_server = os.environ.get("EMAIL_IMAP_SERVER")
required_settings = [
self.test_imap_server,
@ -35,8 +33,7 @@ class TestImap(EmailMessageTestCase):
)
self.mailbox = Mailbox.objects.create(
name='Integration Test Imap',
uri=self.get_connection_string()
name="Integration Test Imap", uri=self.get_connection_string()
)
self.arbitrary_identifier = str(uuid.uuid4())
@ -48,20 +45,19 @@ class TestImap(EmailMessageTestCase):
)
def test_get_imap_message(self):
text_content = 'This is some content'
text_content = "This is some content"
msg = EmailMultiAlternatives(
self.arbitrary_identifier,
text_content,
self.test_from_email,
[
self.test_account,
]
],
)
msg.send()
messages = self._get_new_messages(
self.mailbox,
condition=lambda m: m['subject'] == self.arbitrary_identifier
self.mailbox, condition=lambda m: m["subject"] == self.arbitrary_identifier
)
message = next(messages)

View file

@ -5,15 +5,15 @@ from django.test import TestCase
from django_mailbox.models import Mailbox
__all__ = ['TestMailbox']
__all__ = ["TestMailbox"]
class TestMailbox(TestCase):
def test_protocol_info(self):
mailbox = Mailbox()
mailbox.uri = 'alpha://test.com'
mailbox.uri = "alpha://test.com"
expected_protocol = 'alpha'
expected_protocol = "alpha"
actual_protocol = mailbox._protocol_info.scheme
self.assertEqual(
@ -23,24 +23,30 @@ class TestMailbox(TestCase):
def test_last_polling_field_exists(self):
mailbox = Mailbox()
self.assertTrue(hasattr(mailbox, 'last_polling'))
self.assertTrue(hasattr(mailbox, "last_polling"))
def test_get_new_mail_update_last_polling(self):
mailbox = Mailbox.objects.create(uri="mbox://" + os.path.join(
mailbox = Mailbox.objects.create(
uri="mbox://"
+ os.path.join(
os.path.dirname(__file__),
'messages',
'generic_message.eml',
))
"messages",
"generic_message.eml",
)
)
self.assertEqual(mailbox.last_polling, None)
list(mailbox.get_new_mail())
self.assertNotEqual(mailbox.last_polling, None)
def test_queryset_get_new_mail(self):
mailbox = Mailbox.objects.create(uri="mbox://" + os.path.join(
mailbox = Mailbox.objects.create(
uri="mbox://"
+ os.path.join(
os.path.dirname(__file__),
'messages',
'generic_message.eml',
))
"messages",
"generic_message.eml",
)
)
Mailbox.objects.filter(pk=mailbox.pk).get_new_mail()
mailbox.refresh_from_db()
self.assertNotEqual(mailbox.last_polling, None)

View file

@ -7,20 +7,20 @@ from django_mailbox.models import Message
from django_mailbox.tests.base import EmailMessageTestCase
__all__ = ['TestMessageFlattening']
__all__ = ["TestMessageFlattening"]
class TestMessageFlattening(EmailMessageTestCase):
def test_quopri_message_is_properly_rehydrated(self):
incoming_email_object = self._get_email_object(
'message_with_many_multiparts.eml',
"message_with_many_multiparts.eml",
)
# Note: this is identical to the above, but it appears that
# while reading-in an e-mail message, we do alter it slightly
expected_email_object = self._get_email_object(
'message_with_many_multiparts.eml',
"message_with_many_multiparts.eml",
)
models.TEXT_STORED_MIMETYPES = ['text/plain']
models.TEXT_STORED_MIMETYPES = ["text/plain"]
msg = self.mailbox.process_incoming_message(incoming_email_object)
@ -33,12 +33,12 @@ class TestMessageFlattening(EmailMessageTestCase):
def test_base64_message_is_properly_rehydrated(self):
incoming_email_object = self._get_email_object(
'message_with_attachment.eml',
"message_with_attachment.eml",
)
# Note: this is identical to the above, but it appears that
# while reading-in an e-mail message, we do alter it slightly
expected_email_object = self._get_email_object(
'message_with_attachment.eml',
"message_with_attachment.eml",
)
msg = self.mailbox.process_incoming_message(incoming_email_object)
@ -52,10 +52,10 @@ class TestMessageFlattening(EmailMessageTestCase):
def test_message_handles_rehydration_problems(self):
incoming_email_object = self._get_email_object(
'message_with_defective_attachment_association.eml',
"message_with_defective_attachment_association.eml",
)
expected_email_object = self._get_email_object(
'message_with_defective_attachment_association_result.eml',
"message_with_defective_attachment_association_result.eml",
)
# Note: this is identical to the above, but it appears that
# while reading-in an e-mail message, we do alter it slightly
@ -74,17 +74,17 @@ class TestMessageFlattening(EmailMessageTestCase):
def test_message_content_type_stripping(self):
incoming_email_object = self._get_email_object(
'message_with_many_multiparts.eml',
"message_with_many_multiparts.eml",
)
expected_email_object = self._get_email_object(
'message_with_many_multiparts_stripped_html.eml',
"message_with_many_multiparts_stripped_html.eml",
)
default_settings = utils.get_settings()
with mock.patch('django_mailbox.utils.get_settings') as get_settings:
with mock.patch("django_mailbox.utils.get_settings") as get_settings:
altered = copy.deepcopy(default_settings)
altered['strip_unallowed_mimetypes'] = True
altered['allowed_mimetypes'] = ['text/plain']
altered["strip_unallowed_mimetypes"] = True
altered["allowed_mimetypes"] = ["text/plain"]
get_settings.return_value = altered
@ -100,7 +100,7 @@ class TestMessageFlattening(EmailMessageTestCase):
def test_message_processing_unknown_encoding(self):
incoming_email_object = self._get_email_object(
'message_with_invalid_encoding.eml',
"message_with_invalid_encoding.eml",
)
msg = self.mailbox.process_incoming_message(incoming_email_object)

View file

@ -1,6 +1,7 @@
import gzip
import os.path
import sys
# import sys
import copy
from unittest import mock
@ -12,64 +13,56 @@ from django_mailbox.tests.base import EmailMessageTestCase
from django.utils.encoding import force_text
from django.core.mail import EmailMessage
__all__ = ['TestProcessEmail']
__all__ = ["TestProcessEmail"]
class TestProcessEmail(EmailMessageTestCase):
def test_message_without_attachments(self):
message = self._get_email_object('generic_message.eml')
message = self._get_email_object("generic_message.eml")
mailbox = Mailbox.objects.create()
msg = mailbox.process_incoming_message(message)
self.assertEqual(
msg.mailbox,
mailbox
)
self.assertEqual(msg.subject, 'Message Without Attachment')
self.assertEqual(msg.mailbox, mailbox)
self.assertEqual(msg.subject, "Message Without Attachment")
self.assertEqual(
msg.message_id,
(
'<CAMdmm+hGH8Dgn-_0xnXJCd=PhyNAiouOYm5zFP0z'
'-foqTO60zA@mail.gmail.com>'
)
("<CAMdmm+hGH8Dgn-_0xnXJCd=PhyNAiouOYm5zFP0z" "-foqTO60zA@mail.gmail.com>"),
)
self.assertEqual(
msg.from_header,
'Adam Coddington <test@adamcoddington.net>',
"Adam Coddington <test@adamcoddington.net>",
)
self.assertEqual(
msg.to_header,
'Adam Coddington <test@adamcoddington.net>',
"Adam Coddington <test@adamcoddington.net>",
)
def test_message_with_encoded_attachment_filenames(self):
message = self._get_email_object(
'message_with_koi8r_filename_attachments.eml'
)
message = self._get_email_object("message_with_koi8r_filename_attachments.eml")
mailbox = Mailbox.objects.create()
msg = mailbox.process_incoming_message(message)
attachments = msg.attachments.order_by('pk').all()
attachments = msg.attachments.order_by("pk").all()
self.assertEqual(
'\u041f\u0430\u043a\u0435\u0442 \u043f\u0440\u0435\u0434\u043b'
'\u043e\u0436\u0435\u043d\u0438\u0439 HSE Career Fair 8 \u0430'
'\u043f\u0440\u0435\u043b\u044f 2016.pdf',
attachments[0].get_filename()
"\u041f\u0430\u043a\u0435\u0442 \u043f\u0440\u0435\u0434\u043b"
"\u043e\u0436\u0435\u043d\u0438\u0439 HSE Career Fair 8 \u0430"
"\u043f\u0440\u0435\u043b\u044f 2016.pdf",
attachments[0].get_filename(),
)
self.assertEqual(
'\u0412\u0435\u0434\u043e\u043c\u043e\u0441\u0442\u0438.pdf',
attachments[1].get_filename()
"\u0412\u0435\u0434\u043e\u043c\u043e\u0441\u0442\u0438.pdf",
attachments[1].get_filename(),
)
self.assertEqual(
'\u041f\u0430\u043a\u0435\u0442 \u043f\u0440\u0435\u0434\u043b'
'\u043e\u0436\u0435\u043d\u0438\u0439 2016.pptx',
attachments[2].get_filename()
"\u041f\u0430\u043a\u0435\u0442 \u043f\u0440\u0435\u0434\u043b"
"\u043e\u0436\u0435\u043d\u0438\u0439 2016.pptx",
attachments[2].get_filename(),
)
def test_message_with_attachments(self):
message = self._get_email_object('message_with_attachment.eml')
message = self._get_email_object("message_with_attachment.eml")
mailbox = Mailbox.objects.create()
msg = mailbox.process_incoming_message(message)
@ -85,16 +78,16 @@ class TestProcessEmail(EmailMessageTestCase):
attachment = msg.attachments.all()[0]
self.assertEqual(
attachment.get_filename(),
'heart.png',
"heart.png",
)
def test_message_with_utf8_attachment_header(self):
""" Ensure that we properly handle UTF-8 encoded attachment
"""Ensure that we properly handle UTF-8 encoded attachment
Safe for regress of #104 too
"""
email_object = self._get_email_object(
'message_with_utf8_attachment.eml',
"message_with_utf8_attachment.eml",
)
mailbox = Mailbox.objects.create()
msg = mailbox.process_incoming_message(email_object)
@ -108,24 +101,18 @@ class TestProcessEmail(EmailMessageTestCase):
)
attachment = msg.attachments.all()[0]
self.assertEqual(
attachment.get_filename(),
'pi\u0142kochwyty.jpg'
)
self.assertEqual(attachment.get_filename(), "pi\u0142kochwyty.jpg")
attachment = msg.attachments.all()[1]
self.assertEqual(
attachment.get_filename(),
'odpowied\u017a Burmistrza.jpg'
)
self.assertEqual(attachment.get_filename(), "odpowied\u017a Burmistrza.jpg")
def test_message_get_text_body(self):
message = self._get_email_object('multipart_text.eml')
message = self._get_email_object("multipart_text.eml")
mailbox = Mailbox.objects.create()
msg = mailbox.process_incoming_message(message)
expected_results = 'Hello there!'
expected_results = "Hello there!"
actual_results = msg.text.strip()
self.assertEqual(
@ -135,29 +122,21 @@ class TestProcessEmail(EmailMessageTestCase):
def test_get_text_body_properly_recomposes_line_continuations(self):
message = Message()
email_object = self._get_email_object(
'message_with_long_text_lines.eml'
)
email_object = self._get_email_object("message_with_long_text_lines.eml")
message.get_email_object = lambda: email_object
actual_text = message.text
expected_text = (
'The one of us with a bike pump is far ahead, '
'but a man stopped to help us and gave us his pump.'
"The one of us with a bike pump is far ahead, "
"but a man stopped to help us and gave us his pump."
)
self.assertEqual(
actual_text,
expected_text
)
self.assertEqual(actual_text, expected_text)
def test_get_body_properly_handles_unicode_body(self):
with open(
os.path.join(
os.path.dirname(__file__),
'messages/generic_message.eml'
)
os.path.join(os.path.dirname(__file__), "messages/generic_message.eml")
) as f:
unicode_body = f.read()
@ -167,23 +146,18 @@ class TestProcessEmail(EmailMessageTestCase):
expected_body = unicode_body
actual_body = message.get_email_object().as_string()
self.assertEqual(
expected_body,
actual_body
)
self.assertEqual(expected_body, actual_body)
def test_message_issue_82(self):
""" Ensure that we properly handle incorrectly encoded messages
"""
email_object = self._get_email_object('email_issue_82.eml')
it = 'works'
"""Ensure that we properly handle incorrectly encoded messages"""
email_object = self._get_email_object("email_issue_82.eml")
it = "works"
try:
# it's ok to call as_string() before passing email_object
# to _get_dehydrated_message()
email_object.as_string()
except:
it = 'do not works'
except Exception:
it = "do not works"
success = True
try:
@ -191,15 +165,15 @@ class TestProcessEmail(EmailMessageTestCase):
except ValueError:
success = False
self.assertEqual(it, 'works')
self.assertEqual(it, "works")
self.assertEqual(True, success)
def test_message_issue_82_bis(self):
""" Ensure that the email object is good before and after
"""Ensure that the email object is good before and after
calling _get_dehydrated_message()
"""
message = self._get_email_object('email_issue_82.eml')
message = self._get_email_object("email_issue_82.eml")
success = True
@ -209,16 +183,16 @@ class TestProcessEmail(EmailMessageTestCase):
# msg.eml.save('%s.eml' % uuid.uuid4(), ContentFile(message),
# save=False)
msg.mailbox = self.mailbox
if 'subject' in message:
msg.subject = convert_header_to_unicode(message['subject'])[0:255]
if 'message-id' in message:
msg.message_id = message['message-id'][0:255]
if 'from' in message:
msg.from_header = convert_header_to_unicode(message['from'])
if 'to' in message:
msg.to_header = convert_header_to_unicode(message['to'])
elif 'Delivered-To' in message:
msg.to_header = convert_header_to_unicode(message['Delivered-To'])
if "subject" in message:
msg.subject = convert_header_to_unicode(message["subject"])[0:255]
if "message-id" in message:
msg.message_id = message["message-id"][0:255]
if "from" in message:
msg.from_header = convert_header_to_unicode(message["from"])
if "to" in message:
msg.to_header = convert_header_to_unicode(message["to"])
elif "Delivered-To" in message:
msg.to_header = convert_header_to_unicode(message["Delivered-To"])
msg.save()
# here the message is ok
@ -227,14 +201,14 @@ class TestProcessEmail(EmailMessageTestCase):
try:
# here as_string raises UnicodeEncodeError
str_msg = message.as_string()
except:
except Exception:
success = False
msg.set_body(str_msg)
if message['in-reply-to']:
if message["in-reply-to"]:
try:
msg.in_reply_to = Message.objects.filter(
message_id=message['in-reply-to']
message_id=message["in-reply-to"]
)[0]
except IndexError:
pass
@ -243,18 +217,20 @@ class TestProcessEmail(EmailMessageTestCase):
self.assertEqual(True, success)
def test_message_with_misplaced_utf8_content(self):
""" Ensure that we properly handle incorrectly encoded messages
"""Ensure that we properly handle incorrectly encoded messages
``message_with_utf8_char.eml``'s primary text payload is marked
as being iso-8859-1 data, but actually contains UTF-8 bytes.
"""
email_object = self._get_email_object('message_with_utf8_char.eml')
email_object = self._get_email_object("message_with_utf8_char.eml")
msg = self.mailbox.process_incoming_message(email_object)
expected_text = 'This message contains funny UTF16 characters ' + \
'like this one: "\xc2\xa0" and this one "\xe2\x9c\xbf".'
expected_text = (
"This message contains funny UTF16 characters "
+ 'like this one: "\xc2\xa0" and this one "\xe2\x9c\xbf".'
)
actual_text = msg.text
self.assertEqual(
@ -263,7 +239,7 @@ class TestProcessEmail(EmailMessageTestCase):
)
def test_message_with_invalid_content_for_declared_encoding(self):
""" Ensure that we gracefully handle mis-encoded bodies.
"""Ensure that we gracefully handle mis-encoded bodies.
Should a payload body be misencoded, we should:
@ -275,7 +251,7 @@ class TestProcessEmail(EmailMessageTestCase):
"""
email_object = self._get_email_object(
'message_with_invalid_content_for_declared_encoding.eml',
"message_with_invalid_content_for_declared_encoding.eml",
)
msg = self.mailbox.process_incoming_message(email_object)
@ -284,18 +260,20 @@ class TestProcessEmail(EmailMessageTestCase):
def test_message_with_valid_content_in_single_byte_encoding(self):
email_object = self._get_email_object(
'message_with_single_byte_encoding.eml',
"message_with_single_byte_encoding.eml",
)
msg = self.mailbox.process_incoming_message(email_object)
actual_text = msg.text
expected_body = '\u042d\u0442\u043e ' + \
'\u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 ' + \
'\u0438\u043c\u0435\u0435\u0442 ' + \
'\u043d\u0435\u043f\u0440\u0430\u0432\u0438\u043b\u044c\u043d' + \
'\u0443\u044e ' + \
'\u043a\u043e\u0434\u0438\u0440\u043e\u0432\u043a\u0430.'
expected_body = (
"\u042d\u0442\u043e "
+ "\u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 "
+ "\u0438\u043c\u0435\u0435\u0442 "
+ "\u043d\u0435\u043f\u0440\u0430\u0432\u0438\u043b\u044c\u043d"
+ "\u0443\u044e "
+ "\u043a\u043e\u0434\u0438\u0440\u043e\u0432\u043a\u0430."
)
self.assertEqual(
actual_text,
@ -304,42 +282,44 @@ class TestProcessEmail(EmailMessageTestCase):
def test_message_with_single_byte_subject_encoding(self):
email_object = self._get_email_object(
'message_with_single_byte_extended_subject_encoding.eml',
"message_with_single_byte_extended_subject_encoding.eml",
)
msg = self.mailbox.process_incoming_message(email_object)
expected_subject = '\u00D3\u00E7\u00ED\u00E0\u00E9 \u00EA\u00E0\u00EA ' + \
'\u00E7\u00E0\u00F0\u00E0\u00E1\u00E0\u00F2\u00FB\u00E2' + \
'\u00E0\u00F2\u00FC \u00EE\u00F2 1000$ \u00E2 ' + \
'\u00ED\u00E5\u00E4\u00E5\u00EB\u00FE!'
expected_subject = (
"\u00D3\u00E7\u00ED\u00E0\u00E9 \u00EA\u00E0\u00EA "
+ "\u00E7\u00E0\u00F0\u00E0\u00E1\u00E0\u00F2\u00FB\u00E2"
+ "\u00E0\u00F2\u00FC \u00EE\u00F2 1000$ \u00E2 "
+ "\u00ED\u00E5\u00E4\u00E5\u00EB\u00FE!"
)
actual_subject = msg.subject
self.assertEqual(actual_subject, expected_subject)
expected_from = 'test test <mr.test32@mail.ru>'
expected_from = "test test <mr.test32@mail.ru>"
actual_from = msg.from_header
self.assertEqual(expected_from, actual_from)
def test_message_reply(self):
email_object = EmailMessage(
'Test subject', # subject
'Test body', # body
'username@example.com', # from
['mr.test32@mail.ru'], # to
"Test subject", # subject
"Test body", # body
"username@example.com", # from
["mr.test32@mail.ru"], # to
)
msg = self.mailbox.record_outgoing_message(email_object.message())
self.assertTrue(msg.outgoing)
actual_from = 'username@example.com'
actual_from = "username@example.com"
reply_email_object = EmailMessage(
'Test subject', # subject
'Test body', # body
"Test subject", # subject
"Test body", # body
actual_from, # from
['mr.test32@mail.ru'], # to
["mr.test32@mail.ru"], # to
)
with mock.patch.object(reply_email_object, 'send'):
with mock.patch.object(reply_email_object, "send"):
reply_msg = msg.reply(reply_email_object)
self.assertEqual(reply_msg.in_reply_to, msg)
@ -348,62 +328,58 @@ class TestProcessEmail(EmailMessageTestCase):
reply_email_object.from_email = None
with mock.patch.object(reply_email_object, 'send'):
with mock.patch.object(reply_email_object, "send"):
second_reply_msg = msg.reply(reply_email_object)
self.assertEqual(self.mailbox.from_email, second_reply_msg.from_header)
def test_message_with_text_attachment(self):
email_object = self._get_email_object(
'message_with_text_attachment.eml',
"message_with_text_attachment.eml",
)
msg = self.mailbox.process_incoming_message(email_object)
self.assertEqual(msg.attachments.all().count(), 1)
self.assertEqual('Has an attached text document, too!', msg.text)
self.assertEqual("Has an attached text document, too!", msg.text)
def test_message_with_long_content(self):
email_object = self._get_email_object(
'message_with_long_content.eml',
"message_with_long_content.eml",
)
size = len(force_text(email_object.as_string()))
msg = self.mailbox.process_incoming_message(email_object)
self.assertEqual(size,
len(force_text(msg.get_email_object().as_string())))
self.assertEqual(size, len(force_text(msg.get_email_object().as_string())))
def test_message_saved(self):
message = self._get_email_object('generic_message.eml')
message = self._get_email_object("generic_message.eml")
default_settings = utils.get_settings()
with mock.patch('django_mailbox.utils.get_settings') as get_settings:
with mock.patch("django_mailbox.utils.get_settings") as get_settings:
altered = copy.deepcopy(default_settings)
altered['store_original_message'] = True
altered["store_original_message"] = True
get_settings.return_value = altered
msg = self.mailbox.process_incoming_message(message)
self.assertNotEquals(msg.eml, None)
self.assertTrue(msg.eml.name.endswith('.eml'))
self.assertTrue(msg.eml.name.endswith(".eml"))
with open(msg.eml.name, 'rb') as f:
self.assertEqual(
f.read(),
self._get_email_as_text('generic_message.eml')
)
with open(msg.eml.name, "rb") as f:
self.assertEqual(f.read(), self._get_email_as_text("generic_message.eml"))
def test_message_saving_ignored(self):
message = self._get_email_object('generic_message.eml')
message = self._get_email_object("generic_message.eml")
default_settings = utils.get_settings()
with mock.patch('django_mailbox.utils.get_settings') as get_settings:
with mock.patch("django_mailbox.utils.get_settings") as get_settings:
altered = copy.deepcopy(default_settings)
altered['store_original_message'] = False
altered["store_original_message"] = False
get_settings.return_value = altered
msg = self.mailbox.process_incoming_message(message)
@ -411,22 +387,21 @@ class TestProcessEmail(EmailMessageTestCase):
self.assertEquals(msg.eml, None)
def test_message_compressed(self):
message = self._get_email_object('generic_message.eml')
message = self._get_email_object("generic_message.eml")
default_settings = utils.get_settings()
with mock.patch('django_mailbox.utils.get_settings') as get_settings:
with mock.patch("django_mailbox.utils.get_settings") as get_settings:
altered = copy.deepcopy(default_settings)
altered['compress_original_message'] = True
altered['store_original_message'] = True
altered["compress_original_message"] = True
altered["store_original_message"] = True
get_settings.return_value = altered
msg = self.mailbox.process_incoming_message(message)
actual_email_object = msg.get_email_object()
print(actual_email_object)
self.assertTrue(msg.eml.name.endswith(".eml.gz"))
self.assertTrue(msg.eml.name.endswith('.eml.gz'))
with gzip.open(msg.eml.name, 'rb') as f:
self.assertEqual(f.read(),
self._get_email_as_text('generic_message.eml'))
with gzip.open(msg.eml.name, "rb") as f:
self.assertEqual(f.read(), self._get_email_as_text("generic_message.eml"))

View file

@ -5,23 +5,26 @@ from django.core.management import call_command, CommandError
from django.test import TestCase
import django
class CommandsTestCase(TestCase):
def test_processincomingmessage_no_args(self):
"""Check that processincomingmessage works with no args"""
mailbox_name = None
# Mock handle so that the test doesn't hang waiting for input. Note that we are only testing
# the argument parsing here -- functionality should be tested elsewhere
with mock.patch('django_mailbox.management.commands.processincomingmessage.Command.handle') as handle:
with mock.patch(
"django_mailbox.management.commands.processincomingmessage.Command.handle"
) as handle:
# Don't care about the return value
handle.return_value = None
call_command('processincomingmessage')
call_command("processincomingmessage")
args, kwargs = handle.call_args
# Make sure that we called with the right arguments
try:
self.assertEqual(kwargs['mailbox_name'], mailbox_name)
self.assertEqual(kwargs["mailbox_name"], mailbox_name)
except KeyError:
# Handle Django 1.7
# It uses optparse instead of argparse, so instead of being
@ -29,19 +32,20 @@ class CommandsTestCase(TestCase):
# Thus we expect an empty tuple here
self.assertEqual(args, tuple())
def test_processincomingmessage_with_arg(self):
"""Check that processincomingmessage works with mailbox_name given"""
mailbox_name = 'foo_mailbox'
mailbox_name = "foo_mailbox"
with mock.patch('django_mailbox.management.commands.processincomingmessage.Command.handle') as handle:
with mock.patch(
"django_mailbox.management.commands.processincomingmessage.Command.handle"
) as handle:
handle.return_value = None
call_command('processincomingmessage', mailbox_name)
call_command("processincomingmessage", mailbox_name)
args, kwargs = handle.call_args
try:
self.assertEqual(kwargs['mailbox_name'], mailbox_name)
self.assertEqual(kwargs["mailbox_name"], mailbox_name)
except (AssertionError, KeyError):
# Handle Django 1.7
# It uses optparse instead of argparse, so instead of being
@ -55,11 +59,12 @@ class CommandsTestCase(TestCase):
# error, which means this test is worthless anyway
# For the "compatibility" versions, unexpected arguments aren't handled
# very well, and result in a TypeError
if (LooseVersion(django.get_version()) >= LooseVersion('1.8') and
LooseVersion(django.get_version()) < LooseVersion('1.10')):
if LooseVersion(django.get_version()) >= LooseVersion("1.8") and LooseVersion(
django.get_version()
) < LooseVersion("1.10"):
with self.assertRaises(TypeError):
call_command('processincomingmessage', 'foo_mailbox', 'invalid_arg')
call_command("processincomingmessage", "foo_mailbox", "invalid_arg")
# In 1.10 and later a proper CommandError should be raised
elif LooseVersion(django.get_version()) >= LooseVersion('1.10'):
elif LooseVersion(django.get_version()) >= LooseVersion("1.10"):
with self.assertRaises(CommandError):
call_command('processincomingmessage', 'foo_mailbox', 'invalid_arg')
call_command("processincomingmessage", "foo_mailbox", "invalid_arg")

View file

@ -1,44 +1,32 @@
from unittest import mock
from os.path import exists
from os import remove
from django.test.utils import override_settings
from django_mailbox.tests.base import EmailMessageTestCase, get_email_as_text
from django_mailbox.transports import ImapTransport, Pop3Transport
FAKE_UID_SEARCH_ANSWER = (
'OK',
"OK",
[
b'18 19 20 21 22 23 24 25 26 27 28 29 ' +
b'30 31 32 33 34 35 36 37 38 39 40 41 42 43 44'
]
b"18 19 20 21 22 23 24 25 26 27 28 29 "
+ b"30 31 32 33 34 35 36 37 38 39 40 41 42 43 44"
],
)
FAKE_UID_FETCH_SIZES = (
'OK',
[
b'1 (UID 18 RFC822.SIZE 58070000000)',
b'2 (UID 19 RFC822.SIZE 2593)'
]
"OK",
[b"1 (UID 18 RFC822.SIZE 58070000000)", b"2 (UID 19 RFC822.SIZE 2593)"],
)
FAKE_UID_FETCH_MSG = (
'OK',
"OK",
[
(
b'1 (UID 18 RFC822 {5807}',
get_email_as_text('generic_message.eml')
),
]
)
FAKE_UID_COPY_MSG = (
'OK',
[
b'[COPYUID 1 2 2] (Success)'
]
(b"1 (UID 18 RFC822 {5807}", get_email_as_text("generic_message.eml")),
],
)
FAKE_UID_COPY_MSG = ("OK", [b"[COPYUID 1 2 2] (Success)"])
FAKE_LIST_ARCHIVE_FOLDERS_ANSWERS = (
'OK',
[
b'(\\HasNoChildren \\All) "/" "[Gmail]/All Mail"'
]
"OK",
[b'(\\HasNoChildren \\All) "/" "[Gmail]/All Mail"'],
)
@ -47,14 +35,14 @@ class IMAPTestCase(EmailMessageTestCase):
def imap_server_uid_method(*args):
cmd = args[0]
arg2 = args[2]
if cmd == 'search':
if cmd == "search":
return FAKE_UID_SEARCH_ANSWER
if cmd == 'copy':
if cmd == "copy":
return FAKE_UID_COPY_MSG
if cmd == 'fetch':
if arg2 == '(RFC822.SIZE)':
if cmd == "fetch":
if arg2 == "(RFC822.SIZE)":
return FAKE_UID_FETCH_SIZES
if arg2 == '(RFC822)':
if arg2 == "(RFC822)":
return FAKE_UID_FETCH_MSG
def imap_server_list_method(pattern=None):
@ -69,39 +57,35 @@ class IMAPTestCase(EmailMessageTestCase):
class TestImapTransport(IMAPTestCase):
def setUp(self):
super().setUp()
self.arbitrary_hostname = 'one.two.three'
self.arbitrary_hostname = "one.two.three"
self.arbitrary_port = 100
self.ssl = False
self.transport = ImapTransport(
self.arbitrary_hostname,
self.arbitrary_port,
self.ssl
self.arbitrary_hostname, self.arbitrary_port, self.ssl
)
self.transport.server = self.imap_server
def test_get_email_message(self):
if exists("store.json"):
remove("store.json")
actual_messages = list(self.transport.get_message())
self.assertEqual(len(actual_messages), 27)
actual_message = actual_messages[0]
expected_message = self._get_email_object('generic_message.eml')
expected_message = self._get_email_object("generic_message.eml")
self.assertEqual(expected_message, actual_message)
class TestImapArchivedTransport(TestImapTransport):
def setUp(self):
super().setUp()
self.archive = 'Archive'
self.archive = "Archive"
self.transport = ImapTransport(
self.arbitrary_hostname,
self.arbitrary_port,
self.ssl,
self.archive
self.arbitrary_hostname, self.arbitrary_port, self.ssl, self.archive
)
self.transport.server = self.imap_server
class TestMaxSizeImapTransport(TestImapTransport):
@override_settings(DJANGO_MAILBOX_MAX_MESSAGE_SIZE=5807)
def setUp(self):
super().setUp()
@ -121,38 +105,40 @@ class TestMaxSizeImapTransport(TestImapTransport):
self.assertEqual(len(small_message_ids), 1)
def test_get_email_message(self):
if exists("store.json"):
remove("store.json")
actual_messages = list(self.transport.get_message())
self.assertEqual(len(actual_messages), 1)
actual_message = actual_messages[0]
expected_message = self._get_email_object('generic_message.eml')
expected_message = self._get_email_object("generic_message.eml")
self.assertEqual(expected_message, actual_message)
class TestPop3Transport(EmailMessageTestCase):
def setUp(self):
self.arbitrary_hostname = 'one.two.three'
self.arbitrary_hostname = "one.two.three"
self.arbitrary_port = 100
self.ssl = False
self.transport = Pop3Transport(
self.arbitrary_hostname,
self.arbitrary_port,
self.ssl
self.arbitrary_hostname, self.arbitrary_port, self.ssl
)
self.transport.server = None
super().setUp()
def test_get_email_message(self):
with mock.patch.object(self.transport, 'server') as server:
if exists("store.json"):
remove("store.json")
with mock.patch.object(self.transport, "server") as server:
# Consider this value arbitrary, the second parameter
# should have one entry per message in the inbox
server.list.return_value = [None, ['some_msg']]
server.list.return_value = [None, ["some_msg"]]
server.retr.return_value = [
'+OK message follows',
"+OK message follows",
[
line.encode('ascii')
for line in self._get_email_as_text(
'generic_message.eml'
).decode('ascii').split('\n')
line.encode("ascii")
for line in self._get_email_as_text("generic_message.eml")
.decode("ascii")
.split("\n")
],
10018, # Some arbitrary size, ideally matching the above
]
@ -162,6 +148,6 @@ class TestPop3Transport(EmailMessageTestCase):
self.assertEqual(len(actual_messages), 1)
actual_message = actual_messages[0]
expected_message = self._get_email_object('generic_message.eml')
expected_message = self._get_email_object("generic_message.eml")
self.assertEqual(expected_message, actual_message)