|
| 1 | +import os |
| 2 | +import socket |
| 3 | +import tempfile |
| 4 | + |
| 5 | +from pathlib import Path |
| 6 | + |
| 7 | +from aiosmtpd.controller import Controller |
| 8 | +from django.core.management import call_command |
| 9 | +from django.test import TestCase, override_settings |
| 10 | +from post_office import mail |
| 11 | +from post_office.models import STATUS, Email |
| 12 | + |
| 13 | + |
| 14 | +class _InMemorySMTPHandler: |
| 15 | + def __init__(self): |
| 16 | + self.envelopes = [] |
| 17 | + |
| 18 | + async def handle_DATA(self, server, session, envelope): # noqa: N802 the name is required by aiosmtpd |
| 19 | + self.envelopes.append(envelope) |
| 20 | + return "250 OK" |
| 21 | + |
| 22 | + |
| 23 | +def _allocate_port(): |
| 24 | + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 25 | + sock.bind(("127.0.0.1", 0)) |
| 26 | + try: |
| 27 | + return sock.getsockname()[1] |
| 28 | + finally: |
| 29 | + sock.close() |
| 30 | + |
| 31 | + |
| 32 | +class PostOfficeSMTPIntegrationTest(TestCase): |
| 33 | + def setUp(self): |
| 34 | + self.smtp_handler = _InMemorySMTPHandler() |
| 35 | + self.smtp_port = _allocate_port() |
| 36 | + self.smtp_controller = Controller(self.smtp_handler, hostname="127.0.0.1", port=self.smtp_port) |
| 37 | + self.smtp_controller.start() |
| 38 | + self.addCleanup(self.smtp_controller.stop) |
| 39 | + |
| 40 | + def test_queued_email_is_delivered_via_local_debug_server(self): |
| 41 | + recipients = [ "[email protected]"] |
| 42 | + |
| 43 | + subject = "Queue smoke test" |
| 44 | + body = "Hello from the queue" |
| 45 | + |
| 46 | + with override_settings( |
| 47 | + EMAIL_BACKEND="post_office.EmailBackend", |
| 48 | + EMAIL_HOST="127.0.0.1", |
| 49 | + EMAIL_PORT=self.smtp_port, |
| 50 | + EMAIL_USE_TLS=False, |
| 51 | + EMAIL_HOST_USER="", |
| 52 | + EMAIL_HOST_PASSWORD="", |
| 53 | + ): |
| 54 | + mail.send(recipients=recipients, sender=sender, subject=subject, message=body) |
| 55 | + |
| 56 | + self.assertEqual(Email.objects.filter(status=STATUS.queued).count(), 1) |
| 57 | + |
| 58 | + fd, lockfile_path = tempfile.mkstemp() |
| 59 | + os.close(fd) |
| 60 | + try: |
| 61 | + call_command("send_queued_mail", processes=1, lockfile=lockfile_path, verbosity=0) |
| 62 | + finally: |
| 63 | + Path(lockfile_path).unlink() |
| 64 | + |
| 65 | + sent_emails = Email.objects.filter(status=STATUS.sent) |
| 66 | + self.assertEqual(sent_emails.count(), 1) |
| 67 | + self.assertEqual(len(self.smtp_handler.envelopes), 1) |
| 68 | + envelope = self.smtp_handler.envelopes[0] |
| 69 | + self.assertEqual(envelope.mail_from, sender) |
| 70 | + self.assertEqual(envelope.rcpt_tos, recipients) |
| 71 | + self.assertIn(subject, envelope.content.decode()) |
0 commit comments