Authorization results not yet available for Mailbox {{ mbx.id }}.
+{% endif %} + +{% endblock %} diff --git a/example/templates/mailbox/base.html b/example/templates/mailbox/base.html new file mode 100644 index 0000000..6f9c111 --- /dev/null +++ b/example/templates/mailbox/base.html @@ -0,0 +1,30 @@ + + + + {% load static %} + +| ID | +Name | +From Email | +URI | +Active | +Last_polling | +Action | +|
|---|---|---|---|---|---|---|---|
| {{ mbx.id }} | +{{ mbx.name }} | +{{ mbx.from_email }} | +{{ mbx.uri }} | +{{ mbx.active }} | +{{ mbx.last_polling }} | +Authorize w/mbx | +Authorize w/Google + | +
No Mailboxes are configured yet.
+{% endif %} + +{% endblock %} diff --git a/example/test_connections.py b/example/test_connections.py new file mode 100644 index 0000000..0731b3f --- /dev/null +++ b/example/test_connections.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python +""" +Example script demonstrating how to test Mailbox and Outbox connections +using the new test_connection methods. + +This script can be run from the command line to test connections for +all configured Mailboxes and Outboxes, or specific ones by ID. +""" + +import os +import sys +import django +import argparse + +# Set up Django environment +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) +os.environ.setdefault("DJANGO_SETTINGS_MODULE", "example.settings") +django.setup() + +from django_mail_admin.models import Mailbox, Outbox + + +def test_mailbox_connection(mailbox_id=None): + """ + Test connection for a specific mailbox or all mailboxes. + + Args: + mailbox_id (int, optional): ID of the mailbox to test. If None, test all mailboxes. + """ + if mailbox_id: + try: + mailboxes = [Mailbox.objects.get(id=mailbox_id)] + except Mailbox.DoesNotExist: + print(f"Mailbox with ID {mailbox_id} does not exist.") + return + else: + mailboxes = Mailbox.objects.all() + + if not mailboxes: + print("No mailboxes found.") + return + + for mailbox in mailboxes: + print(f"Testing connection for Mailbox: {mailbox.name} (ID: {mailbox.id})") + success, message = mailbox.test_connection() + status = "SUCCESS" if success else "FAILED" + print(f" Status: {status}") + print(f" Message: {message}") + print() + + +def test_outbox_connection(outbox_id=None): + """ + Test connection for a specific outbox or all outboxes. + + Args: + outbox_id (int, optional): ID of the outbox to test. If None, test all outboxes. + """ + if outbox_id: + try: + outboxes = [Outbox.objects.get(id=outbox_id)] + except Outbox.DoesNotExist: + print(f"Outbox with ID {outbox_id} does not exist.") + return + else: + outboxes = Outbox.objects.all() + + if not outboxes: + print("No outboxes found.") + return + + for outbox in outboxes: + print(f"Testing connection for Outbox: {outbox.name} (ID: {outbox.id})") + success, message = outbox.test_connection() + status = "SUCCESS" if success else "FAILED" + print(f" Status: {status}") + print(f" Message: {message}") + print() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Test email connections for django_mail_admin." + ) + parser.add_argument( + "--mailbox", type=int, help="ID of the mailbox to test", default=None + ) + parser.add_argument( + "--outbox", type=int, help="ID of the outbox to test", default=None + ) + parser.add_argument( + "--all", action="store_true", help="Test all mailboxes and outboxes" + ) + + args = parser.parse_args() + + if args.mailbox: + test_mailbox_connection(args.mailbox) + elif args.outbox: + test_outbox_connection(args.outbox) + elif args.all: + print("Testing all mailboxes:") + test_mailbox_connection() + print("\nTesting all outboxes:") + test_outbox_connection() + else: + parser.print_help() diff --git a/requirements.txt b/requirements.txt index 8664816..50006d5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,8 @@ jsonfield # Additional requirements go here social-auth-app-django +asgiref==3.8.0 +O365 +azure-core==1.30.1 +azure-storage-blob==12.19.1 +dateparser==1.2.0 \ No newline at end of file diff --git a/requirements_test.txt b/requirements_test.txt index 45a130a..a428bb1 100644 --- a/requirements_test.txt +++ b/requirements_test.txt @@ -6,3 +6,4 @@ codecov>=2.0.0 mock==2.0.0 # Additional test requirements go here +python-decouple==3.8 diff --git a/tests/o365/__init__.py b/tests/o365/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/o365/test_commands.py b/tests/o365/test_commands.py new file mode 100644 index 0000000..7dfde3e --- /dev/null +++ b/tests/o365/test_commands.py @@ -0,0 +1,151 @@ +import time +import random + +from urllib import parse +from datetime import datetime + +from django.test import TestCase +from django.conf import settings +from django.template.loader import render_to_string + +from django_mail_admin.models import Outbox, IncomingEmail, OutgoingEmail, STATUS +from django_mail_admin.models import Mailbox +from django_mail_admin.mail import send_queued +from django_mail_admin.connections import connections + +from django_mail_admin.o365_utils import O365NotAuthenticated + + +class O365CommandTest(TestCase): + O365_BACKEND_ALIAS = "o365" + + def print_incoming_emails(self): + for email in IncomingEmail.objects.all().order_by("processed"): + print(f"{email.message_id} | {email.subject} | {email.processed}") + + def _get_html_draft( + self, + text_body: str, + message_hash: str, + legal_disclaimer="This is a test legal disclaimer", + ): + context = { + "bodylines": text_body.splitlines(), + "message_hash": message_hash, + "legal_disclaimer": legal_disclaimer, + } + return render_to_string("html_email_draft.html", context) + + def send_email(self, outgoing_email) -> bool: + retry_send = False + send_successful = False + while True: + try: + outgoing_email.dispatch(commit=True) + except O365NotAuthenticated as uae: + connection = connections[self.O365_BACKEND_ALIAS] + print(f"warning: send_mail: {uae};") + if not retry_send: + retry_send = connection.conn.account.authenticate() + print( + f"authentication attempt: " + + ("successful " if retry_send else " failed!") + ) + else: + retry_send = False + except Exception as e: + print(f"error: send_mail: {e}") + else: + send_successful = True + retry_send = False + + if send_successful or not retry_send: + break + return send_successful + + def test_o365(self): + """ + An end-to-end test of sending one email from the mailbox of user1, and checking if it is received in the mailbox of configured user2, if no user2 is configured, sends it to user1. + + It is likely to through O365NotAuthenticated if the mailboxe authorizations are not already setup for user1 and user2. + """ + from_user_email = settings.O365_TEST_ADMIN.get("test_from_email") + to_user_email = settings.O365_TEST_ADMIN.get("test_to_email", from_user_email) + + from_user_email_str = parse.quote(from_user_email) + from_user_o365_con = f"office365://{from_user_email_str}@outlook.office365.com?client_app_id=test_webapp1" + + Outbox.objects.create( + name="O365CommandTest_Outbox", + email_host=from_user_o365_con, + email_host_user=from_user_email, + email_host_password="ase123hgfd", + active=True, + ) + + test_subject = f"UnitTest Subject Dated {datetime.now()}" + message_hash = random.random() + # do not use '\n' sequence in test body, the returning email will collapse all new line sequences to a single space + text_body = f"UnitTest Body. Line1 Hi\nLine2 This is a test draft" + test_body = f"{text_body}\n{message_hash}" + test_html_body = self._get_html_draft(text_body, message_hash) + """ + print(f"\ntest_subject: {test_subject}") + print(f"test_body: {test_body}\n") + """ + outgoing_emails = [] + for tst_body in [test_body, test_html_body]: + outgoing_emails.append( + OutgoingEmail.objects.create( + from_email=from_user_email, + to=[to_user_email], + status=STATUS.queued, + subject=test_subject, + message=tst_body, + backend_alias=self.O365_BACKEND_ALIAS, + ) + ) + + to_user_email_str = parse.quote(to_user_email) + to_user_o365_con = f"office365://{to_user_email_str}@outlook.office365.com?client_app_id=test_webapp1" + + inbox = Mailbox.objects.create( + name="O365CommandTest_Inbox", uri=to_user_o365_con, from_email=to_user_email + ) + # first poll for inbox emails of user2 + new_emails = inbox.get_new_mail() + print(f"received {len(new_emails)} new emails") + # time.sleep(10) + + # send new email(s) from user1 to user2 + for outgoing_email in outgoing_emails: + send_successful = self.send_email(outgoing_email) + + assert send_successful == True + # sleep for sync + # time.sleep(10) + + # get inbox emails for user2 hoping to have received the recently sent email. + max_attempts = 3 + # if the recent attempt did not succeed in receiving the new email, try again for max_attempts times to fetch new emails. + latest_emails = [] + while max_attempts: + max_attempts -= 1 + time.sleep(2) + new_emails = inbox.get_new_mail() + print(f"received {len(new_emails)} new emails") + latest_emails.extend(new_emails) + if len(latest_emails) == 2: + break + print( + f"\tContinue seeking new mails; retry attempts remaining {max_attempts}" + ) + + assert latest_emails + test_body_wo_newlines = test_body.replace("\n", " ") + for latest_email in latest_emails: + assert latest_email.subject == test_subject + body_matches = (latest_email.text == test_body_wo_newlines) or ( + str(message_hash) in latest_email.html + ) + assert body_matches diff --git a/tests/settings.py b/tests/settings.py index f3c3bc9..6dd8b04 100644 --- a/tests/settings.py +++ b/tests/settings.py @@ -4,9 +4,14 @@ import django import os +from decouple import config as _config + DEBUG = True USE_TZ = True +def config(key): + return _config(key, default="") + # Build paths inside the project like this: os.path.join(BASE_DIR, ...) BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) @@ -16,45 +21,84 @@ DATABASES = { "default": { "ENGINE": "django.db.backends.sqlite3", - 'NAME': os.path.join(os.path.dirname(__file__), 'test.db'), - 'TEST_NAME': os.path.join(os.path.dirname(__file__), 'test.db'), + "NAME": os.path.join(os.path.dirname(__file__), "test.db"), + "TEST_NAME": os.path.join(os.path.dirname(__file__), "test.db"), } } CACHES = { - 'default': { - 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache', - 'TIMEOUT': 36000, - 'KEY_PREFIX': 'django_mail_admin', + "default": { + "BACKEND": "django.core.cache.backends.locmem.LocMemCache", + "TIMEOUT": 36000, + "KEY_PREFIX": "django_mail_admin", + }, + "django_mail_admin": { + "BACKEND": "django.core.cache.backends.locmem.LocMemCache", + "TIMEOUT": 36000, + "KEY_PREFIX": "django_mail_admin", }, - 'django_mail_admin': { - 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache', - 'TIMEOUT': 36000, - 'KEY_PREFIX': 'django_mail_admin', - } } DJANGO_MAIL_ADMIN = { - 'BACKENDS': { - 'default': 'django.core.mail.backends.dummy.EmailBackend', - 'locmem': 'django.core.mail.backends.locmem.EmailBackend', - 'error': 'tests.test_backends.ErrorRaisingBackend', - 'smtp': 'django.core.mail.backends.smtp.EmailBackend', - 'connection_tester': 'django_mail_admin.tests.test_mail.ConnectionTestingBackend', - 'custom': 'django_mail_admin.backends.CustomEmailBackend' + "BACKENDS": { + "default": "django.core.mail.backends.dummy.EmailBackend", + "locmem": "django.core.mail.backends.locmem.EmailBackend", + "error": "tests.test_backends.ErrorRaisingBackend", + "smtp": "django.core.mail.backends.smtp.EmailBackend", + "connection_tester": "django_mail_admin.tests.test_mail.ConnectionTestingBackend", + "custom": "django_mail_admin.backends.CustomEmailBackend", + "o365": "django_mail_admin.backends.O365Backend", } } +O365_ADMIN_SETTINGS = { + "TOKEN_BACKEND": "FileSystemTokenBackend", +} + +O365_CLIENT_APP_SETTINGS = { + "test_webapp1": { + # TBD certificate support + "auth": "token_secret", #'certificate' + "TOKEN_BACKEND": "AZBlobStorageTokenBackend", + "O365_CLIENT_ID": config("O365_CLIENT_ID"), + "O365_CLIENT_SECRET": config("O365_CLIENT_SECRET"), + }, +} + +O365_TOKEN_BACKENDS = { + "FileSystemTokenBackend": { + "O365_AUTH_BACKEND_TOKEN_DIR": config("O365_AUTH_BACKEND_TOKEN_DIR"), + "O365_AUTH_BACKEND_TOKEN_FILE": config("O365_AUTH_BACKEND_TOKEN_FILE"), + }, + "AZBlobStorageTokenBackend": { + "O365_AUTH_BACKEND_AZ_BLOB_NAME": config("O365_AUTH_BACKEND_AZ_BLOB_NAME"), + "O365_AUTH_BACKEND_AZ_CONTAINER_PATH": config( + "O365_AUTH_BACKEND_AZ_CONTAINER_PATH" + ), + "O365_AUTH_BACKEND_AZ_CONNECTION_STR": config( + "O365_AUTH_BACKEND_AZ_CONNECTION_STR" + ), + }, +} + +O365_TEST_ADMIN = { + "test_from_email": config("O365_TEST_FROM_ACCOUNT"), + "test_to_email": config("O365_TEST_TO_ACCOUNT"), +} + TEMPLATES = [ { - 'BACKEND': 'django.template.backends.django.DjangoTemplates', - 'DIRS': [os.path.join(BASE_DIR, 'templates'), ], - 'APP_DIRS': True, - 'OPTIONS': { - 'context_processors': [ - 'django.template.context_processors.debug', - 'django.template.context_processors.request', - 'django.contrib.auth.context_processors.auth', - 'django.contrib.messages.context_processors.messages', + "BACKEND": "django.template.backends.django.DjangoTemplates", + "DIRS": [ + os.path.join(BASE_DIR, "templates"), + os.path.join(BASE_DIR, "tests/templates"), + ], + "APP_DIRS": True, + "OPTIONS": { + "context_processors": [ + "django.template.context_processors.debug", + "django.template.context_processors.request", + "django.contrib.auth.context_processors.auth", + "django.contrib.messages.context_processors.messages", ], }, }, diff --git a/tests/templates/html_email_draft.html b/tests/templates/html_email_draft.html new file mode 100644 index 0000000..62d2acf --- /dev/null +++ b/tests/templates/html_email_draft.html @@ -0,0 +1,19 @@ + + + + + + + +
+ {% for bodyline in bodylines %}
+ {{bodyline}}
+ {% endfor %}
+
Kindly retain the following when replying
+
{{ message_hash}}
+
---------------------------------------------------------
+{{legal_disclaimer}}
+ + diff --git a/tests/test_connection_methods.py b/tests/test_connection_methods.py new file mode 100644 index 0000000..1004898 --- /dev/null +++ b/tests/test_connection_methods.py @@ -0,0 +1,96 @@ +import unittest +from unittest.mock import patch, MagicMock + +from django.test import TestCase + +from django_mail_admin.models import Mailbox, Outbox + + +class MailboxConnectionTestCase(TestCase): + """Test the Mailbox.test_connection method""" + + def setUp(self): + self.mailbox = Mailbox.objects.create( + name="Test IMAP Mailbox", uri="imap://user:password@example.com" + ) + + @patch("django_mail_admin.models.configurations.ImapTransport") + def test_imap_connection_success(self, mock_imap): + """Test successful IMAP connection""" + # Set up the mock + mock_connection = MagicMock() + mock_connection.server.noop.return_value = True + mock_imap.return_value = mock_connection + + # Test the connection + success, message = self.mailbox.test_connection() + + # Verify the result + self.assertTrue(success) + self.assertIn("Successfully connected", message) + mock_connection.server.noop.assert_called_once() + + @patch("django_mail_admin.models.configurations.ImapTransport") + def test_imap_connection_failure(self, mock_imap): + """Test failed IMAP connection""" + # Set up the mock to raise an exception + mock_connection = MagicMock() + mock_connection.server.noop.side_effect = Exception("Connection refused") + mock_imap.return_value = mock_connection + + # Test the connection + success, message = self.mailbox.test_connection() + + # Verify the result + self.assertFalse(success) + self.assertIn("Connection failed", message) + self.assertIn("Connection refused", message) + + +class OutboxConnectionTestCase(TestCase): + """Test the Outbox.test_connection method""" + + def setUp(self): + self.outbox = Outbox.objects.create( + name="Test SMTP Outbox", + email_host="smtp.example.com", + email_host_user="user@example.com", + email_host_password="password", + email_port=587, + ) + + @patch("django_mail_admin.models.configurations.connections") + def test_smtp_connection_success(self, mock_connections): + """Test successful SMTP connection""" + # Set up the mock + mock_connection = MagicMock() + mock_connection.connection.noop.return_value = True + mock_connections.__getitem__.return_value = mock_connection + + # Test the connection + success, message = self.outbox.test_connection() + + # Verify the result + self.assertTrue(success) + self.assertIn("Successfully connected", message) + mock_connection.connection.noop.assert_called_once() + + @patch("django_mail_admin.models.configurations.connections") + def test_smtp_connection_failure(self, mock_connections): + """Test failed SMTP connection""" + # Set up the mock to raise an exception + mock_connection = MagicMock() + mock_connection.connection.noop.side_effect = Exception("Authentication failed") + mock_connections.__getitem__.return_value = mock_connection + + # Test the connection + success, message = self.outbox.test_connection() + + # Verify the result + self.assertFalse(success) + self.assertIn("Connection failed", message) + self.assertIn("Authentication failed", message) + + +if __name__ == "__main__": + unittest.main() diff --git a/tox.ini b/tox.ini index 1046e5c..5594631 100644 --- a/tox.ini +++ b/tox.ini @@ -24,7 +24,7 @@ deps = django22: Django>=2.2,<3.0 django30: Django>=3.0,<3.1 django40: Django>=4.0 -passenv=TRAVIS TRAVIS_JOB_ID TRAVIS_BRANCH +passenv=TRAVIS [travis:env] DJANGO = 1.8: django18