diff --git a/repapp/email_interface/tests.py b/repapp/email_interface/tests.py index 0aeed71..6989a5c 100644 --- a/repapp/email_interface/tests.py +++ b/repapp/email_interface/tests.py @@ -1,9 +1,18 @@ +import os import logging -from django.test import TestCase +from pathlib import Path +from unittest.mock import patch +from django.test import TestCase, Client from django.conf import settings +from django.test.client import RequestFactory from django.contrib.auth import get_user_model +from django.core import mail +from django.utils import translation +from django.urls import reverse from .forms import MessageForm from .models import Message, Attachment +from .utils import create_message, send_message, process_message +from one_time_login.utils import create_one_time_login logger = logging.getLogger(__name__) @@ -235,7 +244,7 @@ class MessageModelTestCase(TestCase): class AttachmentModelTestCase(TestCase): """ - Tests all logic contained in the Message model. + Tests all logic contained in the Attachment model. """ def setUp(self): @@ -323,3 +332,508 @@ class AttachmentModelTestCase(TestCase): assert len(Attachment.of_user(self.jane)) == 1 assert Attachment.of_user(self.jane)[0].pk == self.a2.pk + + +class UtilsTestCase(TestCase): + """ + Tests all logic contained in utils. + """ + + def setUp(self): + # prepare request factory + self.factory = RequestFactory() + + # prepare test data + john = get_user_model().objects.create_user( + "john", + "lennon@thebeatles.com", + "johnpassword") + john.save() + self.john = john + + jane = get_user_model().objects.create_user( + "jane", + "doe@example.com", + "janepassword") + jane.save() + self.jane = jane + + m1 = Message( + summary='summary 1', + html_content='

mail 1 HTML content

', + text_content='mail 1 text content', + sender=john, + receiver=jane, + sent=True, + answered=True, + ) + m1.save() + self.m1 = m1 + + self.attachment_file = os.path.join( + Path(__file__).resolve().parent, 'test_data', 'Nut.jpg') + + def test_create_message(self): + request = self.factory.get('/') + sent = create_message( + request=request, + sender=self.john, + receiver=self.jane, + summary='A test message', + html_content='

Test HTML content

', + text_content='Test text content', + attachments=[self.attachment_file, '/does-not-exist'], + reply_to=self.m1, + send=False + ) + + assert not sent + assert len(mail.outbox) == 0 + + message = Message.objects.filter(summary='A test message').first() + + assert message.sender == self.john + assert message.receiver == self.jane + assert message.html_content == '

Test HTML content

' + assert message.text_content == 'Test text content' + assert message.reply_to == self.m1 + + attachments = message.attachments() + + assert len(attachments) == 1 + assert 'Nut' in str(attachments[0].name) + assert 'Nut' in str(attachments[0].file) + assert 'image/jpeg' == attachments[0].mime_type + assert attachments[0].owner == self.john + assert attachments[0].message == message + + path = attachments[0].file.path + + logger.debug('Attachment file: %s', attachments[0].file.path) + + assert os.path.isfile(path) + + message.delete() + # deleting the message should also delete the attachment and the file + assert not os.path.isfile(path) + + def test_send_message(self): + request = self.factory.get('/') + + create_message( + request=request, + sender=self.john, + receiver=self.jane, + summary='Another test message', + html_content=f'', + text_content='Test text content', + attachments=[self.attachment_file, '/does-not-exist'], + reply_to=self.m1, + send=False + ) + + message = Message.objects.filter( + summary='Another test message').first() + + sent = send_message(message=message, request=request) + + assert sent + assert len(mail.outbox) == 1 + + m = mail.outbox[0] + + assert len(m.recipients()) == 1 + assert self.jane.email in m.recipients() + + logger.debug('Mail sender: %r', m.from_email) + + assert m.from_email == settings.DEFAULT_FROM_EMAIL + assert m.body == 'Test text content' + + logger.debug('Mail attachments: %r', len(m.attachments)) + + assert 'Nut' in str(m.attachments[0]) + assert 'image/jpeg' in str(m.attachments[0]) + + absolute_media_url = request.build_absolute_uri(settings.MEDIA_URL) + + assert absolute_media_url in m.message().as_string() + + message.delete() + + def test_create_message_and_send(self): + request = self.factory.get('/') + + sent = create_message( + request=request, + sender=self.john, + receiver=self.jane, + summary='My test message', + html_content='

Test HTML content

', + text_content='Test text content', + attachments=[self.attachment_file, '/does-not-exist'], + reply_to=self.m1, + send=True + ) + + assert sent + assert len(mail.outbox) == 1 + + message = Message.objects.filter(summary='My test message').first() + + message.delete() + + def test_process_message_no_guest(self): + with patch('imap_tools.MailMessage') as MockClass: + mock = MockClass.return_value + + mock.from_ = 'noguest@example.com' + mock.subject = 'subject' + + processed = process_message(mock) + + assert not processed + + def test_process_message_reply(self): + with patch('imap_tools.MailMessage') as MockClass: + mock = MockClass.return_value + + mock.from_ = self.m1.receiver.email + mock.subject = f'Re: {self.m1.summary}' + mock.html = '

HTML content

' + mock.text = 'Text content' + mock.attachments = [] + + processed = process_message(mock) + + assert processed + assert self.m1.answered + + message = Message.objects.filter(summary=mock.subject).first() + + assert message.reply_to == self.m1 + + assert message.sender == self.m1.receiver + assert message.receiver == self.m1.sender + assert message.summary == f'Re: {self.m1.summary}' + assert message.html_content == '

HTML content

' + assert message.text_content == 'Text content' + + def test_process_message_new_message(self): + with patch('imap_tools.MailMessage') as MockClass: + mock = MockClass.return_value + + mock.from_ = self.m1.receiver.email + mock.subject = 'New message' + mock.html = '

HTML content

' + mock.text = 'Text content' + mock.attachments = [] + + processed = process_message(mock) + + assert processed + + message = Message.objects.filter(summary=mock.subject).first() + + assert message.reply_to == None + assert message.sender == self.m1.receiver + assert message.receiver == None + assert message.summary == 'New message' + assert message.html_content == '

HTML content

' + assert message.text_content == 'Text content' + + def test_process_message_attachments(self): + with patch('imap_tools.MailMessage') as MockClass: + with patch('imap_tools.MailAttachment') as MockAttachmentClass: + with open(self.attachment_file, "rb") as file: + mock = MockClass.return_value + attachment_mock = MockAttachmentClass.return_value + + attachment_mock.filename = 'test.jpg' + attachment_mock.content_type = 'image/jpeg' + attachment_mock.payload = file.read() + + mock.from_ = self.m1.receiver.email + mock.subject = 'New message with attachment' + mock.html = '

HTML content

' + mock.text = 'Text content' + mock.attachments = [attachment_mock] + + processed = process_message(mock) + + assert processed + + message = Message.objects.filter( + summary=mock.subject).first() + + assert message.reply_to == None + assert message.sender == self.m1.receiver + assert message.receiver == None + assert message.summary == 'New message with attachment' + assert message.html_content == '

HTML content

' + assert message.text_content == 'Text content' + + attachment = Attachment.objects.filter( + message=message).first() + + logger.debug('Attachment: %r', attachment) + + assert attachment.owner == self.m1.receiver + assert 'test' in attachment.name + assert attachment.mime_type == 'image/jpeg' + assert 'test' in attachment.file.path + assert os.path.isfile(attachment.file.path) + + message.delete() + # deleting the message should also delete the attachment and the file + assert not os.path.isfile(attachment.file.path) + + +class ViewsTestCase(TestCase): + """ + Test email interface views. + """ + + def setUp(self): + # use english translations + translation.activate('en') + + # prepare test data + john = get_user_model().objects.create_user( + "john", + "lennon@thebeatles.com", + "johnpassword") + john.save() + self.john = john + + jane = get_user_model().objects.create_user( + "jane", + "doe@example.com", + "janepassword") + jane.save() + self.jane = jane + + m1 = Message( + summary='summary 1', + html_content='

mail 1 HTML content

', + text_content='mail 1 text content', + sender=john, + receiver=jane, + sent=True, + answered=True, + ) + m1.save() + self.m1 = m1 + + m3 = Message( + summary='summary 3', + html_content='

mail 3 HTML content

', + text_content='mail 3 text content', + sender=jane, + receiver=john, + sent=True, + answered=False, + reply_to=m1, + ) + m3.save() + self.m3 = m3 + + a1 = Attachment( + owner=john, + name='johns_file.txt', + mime_type='text/plain', + file='uploads/johns_file.txt', + message=m1, + ) + a1.save() + self.a1 = a1 + + a2 = Attachment( + owner=jane, + name='janes_image.jpg', + mime_type='image/jpeg', + file='media/janes_image.jpg', + message=m3, + ) + a2.save() + self.a2 = a2 + + a3 = Attachment( + owner=john, + name='johns_image.jpg', + mime_type='image/png', + file='media/johns_image.jpg', + message=m1, + ) + a3.save() + self.a3 = a3 + + def test_my_attachments_is_protected(self): + url = reverse('email_interface:my_attachments') + + client = Client() + + response = client.get(url, follow=True) + logger.debug('test_my_attachments_is_protected: %r', + response.redirect_chain[-1]) + + last_url, status_code = response.redirect_chain[-1] + assert status_code == 302 + assert last_url == '/accounts/login/?next=/en/emails/attachments/' + + def test_my_attachments(self): + user = get_user_model().objects.get(username='john') + + url = reverse('email_interface:my_attachments') + otl = create_one_time_login(user, url) + + client = Client() + + response = client.get(otl.get_absolute_url(), follow=True) + + assert response.status_code == 200, 'response is OK' + last_url, _ = response.redirect_chain[-1] + assert last_url == url, 'redirect URL' + + self.assertContains(response, 'johns_file.txt') + self.assertContains(response, 'johns_image.jpg') + self.assertNotContains(response, 'janes_image') + + def test_my_received_mails_is_protected(self): + url = reverse('email_interface:my_received_mails') + + client = Client() + + response = client.get(url, follow=True) + logger.debug('test_my_received_mails_is_protected: %r', + response.redirect_chain[-1]) + + last_url, status_code = response.redirect_chain[-1] + assert status_code == 302 + assert last_url == '/accounts/login/?next=/en/emails/received/' + + def test_my_received_mails(self): + user = get_user_model().objects.get(username='john') + + url = reverse('email_interface:my_received_mails') + otl = create_one_time_login(user, url) + + client = Client() + + response = client.get(otl.get_absolute_url(), follow=True) + + assert response.status_code == 200, 'response is OK' + last_url, _ = response.redirect_chain[-1] + assert last_url == url, 'redirect URL' + + self.assertContains(response, '

mail 3 HTML content

') + self.assertContains(response, 'janes_image') + self.assertNotContains(response, '

mail 1 HTML content

') + + def test_my_sent_mails_is_protected(self): + url = reverse('email_interface:my_sent_mails') + + client = Client() + + response = client.get(url, follow=True) + logger.debug('test_my_sent_mails_is_protected: %r', + response.redirect_chain[-1]) + + last_url, status_code = response.redirect_chain[-1] + assert status_code == 302 + assert last_url == '/accounts/login/?next=/en/emails/sent/' + + def test_my_sent_mails(self): + user = get_user_model().objects.get(username='john') + + url = reverse('email_interface:my_sent_mails') + otl = create_one_time_login(user, url) + + client = Client() + + response = client.get(otl.get_absolute_url(), follow=True) + + assert response.status_code == 200, 'response is OK' + last_url, _ = response.redirect_chain[-1] + assert last_url == url, 'redirect URL' + + self.assertContains(response, '

mail 1 HTML content

') + self.assertNotContains(response, '

mail 3 HTML content

') + + def test_mail_thread_is_protected(self): + url = reverse('email_interface:mail_thread', kwargs={'id': self.m3.pk}) + + client = Client() + + response = client.get(url, follow=True) + logger.debug('test_my_received_mails_is_protected: %r', + response.redirect_chain[-1]) + + last_url, status_code = response.redirect_chain[-1] + assert status_code == 302 + assert last_url == f'/accounts/login/?next=/en/emails/{self.m3.pk}/view/' + + def test_mail_thread(self): + user = get_user_model().objects.get(username='john') + + url = reverse('email_interface:mail_thread', kwargs={'id': self.m3.pk}) + otl = create_one_time_login(user, url) + + client = Client() + + response = client.get(otl.get_absolute_url(), follow=True) + + assert response.status_code == 200, 'response is OK' + last_url, _ = response.redirect_chain[-1] + assert last_url == url, 'redirect URL' + + self.assertContains(response, '

mail 3 HTML content

') + self.assertContains(response, 'janes_image') + self.assertContains(response, '

mail 1 HTML content

') + + def test_mail_thread_reply(self): + user = get_user_model().objects.get(username='john') + + url = reverse('email_interface:mail_thread', kwargs={'id': self.m3.pk}) + otl = create_one_time_login(user, url) + + client = Client() + + response = client.get(otl.get_absolute_url(), follow=True) + + assert response.status_code == 200, 'response is OK' + last_url, _ = response.redirect_chain[-1] + assert last_url == url, 'redirect URL' + + response = client.post(url, { + 'html_content': '

The reply content.

', + 'receiver': self.jane.pk, + 'use_mail_text': '', + 'mail_text': '', + }, follow=True) + + logger.debug('Post response: %r', response) + + message = Message.objects.filter(reply_to=self.m3).first() + + url = reverse('email_interface:mail_thread', kwargs={'id': message.pk}) + + assert response.status_code == 200, 'response is OK' + last_url, _ = response.redirect_chain[-1] + assert last_url == url, 'redirect URL' + + self.assertContains(response, 'Mail was sent.') + + assert len(mail.outbox) == 1 + + logger.debug('Outbox: %r', mail.outbox[0].subject) + + assert mail.outbox[0].subject == 'Re: summary 3' + assert self.jane.email in mail.outbox[0].recipients() + + assert message.reply_to == self.m3 + assert message.sender == user + assert message.receiver == self.jane + assert message.summary == 'Re: summary 3' + assert message.html_content == '

The reply content.

' + assert 'The reply content.' in message.text_content diff --git a/repapp/email_interface/urls.py b/repapp/email_interface/urls.py index 91623f8..fed7f5e 100644 --- a/repapp/email_interface/urls.py +++ b/repapp/email_interface/urls.py @@ -16,11 +16,11 @@ urlpatterns = [ path("sent/", views.my_sent_mails, name="my_sent_mails"), path("received/", views.my_received_mails, name="my_received_mails"), path("attachments/", views.my_attachments, name="my_attachments"), - path("/view", views.mail_thread, name="mail_thread"), + path("/view/", views.mail_thread, name="mail_thread"), ] -if settings.DEBUG or sys.argv[1:2] == ['test']: - logging.info('email_interface: enable test and debug URLs') +if settings.DEBUG: # pragma: no cover + logging.info('email_interface: enable debug URLs') urlpatterns += [ path("test-mail/", views.send_test_mail, name="send_test_mail"), path("send-mail/", views.SendMailView.as_view(), name="send_mail"), diff --git a/repapp/email_interface/utils.py b/repapp/email_interface/utils.py index d5e8e70..e23107d 100644 --- a/repapp/email_interface/utils.py +++ b/repapp/email_interface/utils.py @@ -35,7 +35,7 @@ def create_message(request, sender, receiver, summary, html_content, text_conten continue _, name = os.path.split(attachment) - mime_type = mimetypes.guess_type(attachment) + mime_type = mimetypes.guess_type(attachment)[0] if not attachment.startswith(settings.MEDIA_ROOT): with open(attachment, "rb") as file: @@ -86,7 +86,8 @@ def send_message(message, request): try: res = email.send() - except: + except: # pragma: no cover + # only happens with broken Django mail config # Sending mail cause an exception return False @@ -95,13 +96,22 @@ def send_message(message, request): message.save() return True - return False + return False # pragma: no cover + # should never happen -def process_message(mail_message: MailMessage, guest): +def process_message(mail_message: MailMessage): """ Process a valid message from a guest. """ + sender = mail_message.from_ + + guest = get_user_model().objects.filter(email=sender, is_active=True).first() + if not guest: + logger.info( + 'Mail form unknown sender %s with subject %s ignored.', sender, mail_message.subject) + return False + receiver = None reply_to = None @@ -160,8 +170,11 @@ def process_message(mail_message: MailMessage, guest): logger.debug('New attachment %s', a) + return True -def process_mails(): + +def process_mails(): # pragma: no cover + # no test for imap_tools needed """ Check for new messages and handle valid messages. """ @@ -175,14 +188,7 @@ def process_mails(): message_count = 0 for message in messages: message_count += 1 - sender = message.from_ - - guest = get_user_model().objects.filter(email=sender, is_active=True).first() - if guest: - process_message(message, guest) - else: - logger.info( - 'Mail form unknown sender %s with subject %s ignored.', sender, message.subject) + process_message(message) logger.debug('Processed %d new messages', message_count) diff --git a/repapp/email_interface/views.py b/repapp/email_interface/views.py index 5efdc9d..34a0298 100644 --- a/repapp/email_interface/views.py +++ b/repapp/email_interface/views.py @@ -18,7 +18,8 @@ from .models import Message, Attachment logger = logging.getLogger(__name__) -def process(request): +def process(request): # pragma: no cover + # no logic contained, no test needed count = process_mails() return HttpResponse(f'{count} new messages were processed.') @@ -42,7 +43,10 @@ def mail_thread(request, id): form.data['receiver'] = message.sender.pk form.data['summary'] = summary + logger.debug('mail_thread: sending mail %r', form) + if form.is_valid(): + logger.debug('mail_thread: mail is valid. %r', form) m = form.save(commit=False) m.sender = request.user m.reply_to = message @@ -56,8 +60,12 @@ def mail_thread(request, id): if success: messages.info(request, _('Mail was sent.')) return HttpResponseRedirect(reverse_lazy('email_interface:mail_thread', kwargs={'id': m.pk})) - else: + else: # pragma: no cover + # should never happen messages.error(request, _('Sending of mail failed!')) + else: # pragma: no cover + # only debug log + logger.debug('mail_thread: mail is not valid! %r', form.errors) return render( request, @@ -73,10 +81,11 @@ def mail_thread(request, id): @login_required -def send_test_mail(request): +def send_test_mail(request): # pragma: no cover """ Debug view to send a test mail. """ + # debug view # get first user user = get_user_model().objects.get(pk=1) current_folder = Path(__file__).resolve().parent @@ -97,7 +106,8 @@ def send_test_mail(request): return HttpResponse('Sending of mail failed!') -class SendMailView(LoginRequiredMixin, generic.FormView): +class SendMailView(LoginRequiredMixin, generic.FormView): # pragma: no cover + # debug view form_class = MessageForm template_name = "email_interface/testing/write_mail.html" success_url = reverse_lazy("email_interface:send_mail")