test for email interface completed

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='<p>mail 1 HTML content</p>',
+            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='<p>Test HTML content</p>',
+            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 == '<p>Test HTML content</p>'
+        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'<img src="{settings.MEDIA_URL}/image.jpg">',
+            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='<p>Test HTML content</p>',
+            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 = '<p>HTML content</p>'
+            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 == '<p>HTML content</p>'
+            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 = '<p>HTML content</p>'
+            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 == '<p>HTML content</p>'
+            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 = '<p>HTML content</p>'
+                    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 == '<p>HTML content</p>'
+                    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='<p>mail 1 HTML content</p>',
+            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='<p>mail 3 HTML content</p>',
+            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, '<p>mail 3 HTML content</p>')
+        self.assertContains(response, 'janes_image')
+        self.assertNotContains(response, '<p>mail 1 HTML content</p>')
+
+    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, '<p>mail 1 HTML content</p>')
+        self.assertNotContains(response, '<p>mail 3 HTML content</p>')
+
+    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, '<p>mail 3 HTML content</p>')
+        self.assertContains(response, 'janes_image')
+        self.assertContains(response, '<p>mail 1 HTML content</p>')
+
+    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': '<p>The reply content.</p>',
+            '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 == '<p>The reply content.</p>'
+        assert 'The reply content.' in message.text_content

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("<int:id>/view", views.mail_thread, name="mail_thread"),
+    path("<int:id>/view/", views.mail_thread, name="mail_thread"),
 ]
 
-if settings.DEBUG or sys.argv[1:2] == ['test']:
+if settings.DEBUG:  # pragma: no cover
-    logging.info('email_interface: enable test and debug URLs')
+    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"),

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_
+        process_message(message)
-
-        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)
 
     logger.debug('Processed %d new messages', message_count)
 

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")