146 lines
5.4 KiB
Python
146 lines
5.4 KiB
Python
"""Unit tests for SMTP mail helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import smtplib
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from media_library_viewer_api.services.mailer import (
|
|
EmailAttachment,
|
|
describe_smtp_error,
|
|
html_to_text,
|
|
send_email_message,
|
|
)
|
|
|
|
|
|
class _SMTPContext:
|
|
def __init__(self, smtp: MagicMock):
|
|
self.smtp = smtp
|
|
|
|
def __enter__(self):
|
|
return self.smtp
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
|
|
class MailerTests(unittest.TestCase):
|
|
def test_html_to_text_strips_tags(self) -> None:
|
|
text = html_to_text("<p>Hello <strong>world</strong></p><p>Line 2</p>")
|
|
self.assertIn("Hello", text)
|
|
self.assertIn("world", text)
|
|
self.assertIn("Line 2", text)
|
|
|
|
def test_send_email_message_uses_smtp_with_attachments(self) -> None:
|
|
settings = SimpleNamespace(
|
|
smtp_host="smtp.example.com",
|
|
smtp_port=587,
|
|
smtp_username="mailer@example.com",
|
|
smtp_password="secret",
|
|
smtp_from_address="mailer@example.com",
|
|
smtp_from_name="Manage",
|
|
smtp_use_tls=True,
|
|
smtp_use_ssl=False,
|
|
smtp_timeout=15,
|
|
)
|
|
smtp = MagicMock()
|
|
smtp.send_message.return_value = {}
|
|
smtp_factory = MagicMock(return_value=_SMTPContext(smtp))
|
|
|
|
with patch("media_library_viewer_api.services.mailer.smtplib.SMTP", smtp_factory), patch(
|
|
"media_library_viewer_api.services.mailer.smtplib.SMTP_SSL"
|
|
) as smtp_ssl:
|
|
result = send_email_message(
|
|
settings,
|
|
recipients=["alex@example.com", "sam@example.com"],
|
|
subject="Hello",
|
|
html_body="<p><strong>Hi</strong> there</p>",
|
|
attachments=[EmailAttachment(filename="note.txt", content_type="text/plain", data=b"note")],
|
|
)
|
|
|
|
smtp_ssl.assert_not_called()
|
|
smtp.starttls.assert_called_once()
|
|
smtp.login.assert_called_once_with("mailer@example.com", "secret")
|
|
smtp.send_message.assert_called_once()
|
|
message = smtp.send_message.call_args.args[0]
|
|
self.assertEqual(message["Subject"], "Hello")
|
|
self.assertEqual(message["From"], "Manage <mailer@example.com>")
|
|
self.assertEqual(result["recipient_count"], 2)
|
|
self.assertEqual(result["attachment_count"], 1)
|
|
def test_send_email_message_falls_back_to_fastmail_mode(self) -> None:
|
|
settings = SimpleNamespace(
|
|
smtp_host="smtp.fastmail.com",
|
|
smtp_port=465,
|
|
smtp_username="mailer@example.com",
|
|
smtp_password="secret",
|
|
smtp_from_address="alias@example.com",
|
|
smtp_from_name="Manage",
|
|
smtp_use_tls=False,
|
|
smtp_use_ssl=True,
|
|
smtp_timeout=15,
|
|
)
|
|
smtp_ssl_factory = MagicMock(side_effect=TimeoutError("timed out"))
|
|
fallback_smtp = MagicMock()
|
|
fallback_factory = MagicMock(return_value=_SMTPContext(fallback_smtp))
|
|
|
|
with patch("media_library_viewer_api.services.mailer.smtplib.SMTP_SSL", smtp_ssl_factory), patch(
|
|
"media_library_viewer_api.services.mailer.smtplib.SMTP",
|
|
fallback_factory,
|
|
):
|
|
result = send_email_message(
|
|
settings,
|
|
recipients=["alex@example.com"],
|
|
subject="Hello",
|
|
html_body="<p>Hello</p>",
|
|
)
|
|
|
|
self.assertEqual(result["selected_mode"]["label"], "Fastmail STARTTLS 587")
|
|
self.assertEqual(result["authenticated_as"], "mailer@example.com")
|
|
self.assertEqual(len(result["attempts"]), 2)
|
|
self.assertEqual(result["attempts"][0]["status"], "failed")
|
|
self.assertEqual(result["attempts"][1]["status"], "ok")
|
|
fallback_smtp.send_message.assert_called_once()
|
|
|
|
def test_send_email_message_rejects_unauthorized_from_address(self) -> None:
|
|
settings = SimpleNamespace(
|
|
smtp_host="smtp.example.com",
|
|
smtp_port=587,
|
|
smtp_username="mailer@example.com",
|
|
smtp_password="secret",
|
|
smtp_from_address="alias@example.com",
|
|
smtp_from_name="Manage",
|
|
smtp_use_tls=True,
|
|
smtp_use_ssl=False,
|
|
smtp_timeout=15,
|
|
)
|
|
smtp = MagicMock()
|
|
smtp.send_message.side_effect = smtplib.SMTPDataError(
|
|
551, b"5.7.1 Not authorised to send from this header address"
|
|
)
|
|
smtp_factory = MagicMock(return_value=_SMTPContext(smtp))
|
|
|
|
with patch("media_library_viewer_api.services.mailer.smtplib.SMTP", smtp_factory), patch(
|
|
"media_library_viewer_api.services.mailer.smtplib.SMTP_SSL"
|
|
) as smtp_ssl:
|
|
with self.assertRaises(RuntimeError) as ctx:
|
|
send_email_message(
|
|
settings,
|
|
recipients=["alex@example.com"],
|
|
subject="Hello",
|
|
html_body="<p>Hello</p>",
|
|
)
|
|
|
|
smtp_ssl.assert_not_called()
|
|
self.assertIn("authorized alias", str(ctx.exception).lower())
|
|
self.assertEqual(smtp.send_message.call_count, 1)
|
|
|
|
def test_describe_smtp_error_handles_timeout(self) -> None:
|
|
detail = describe_smtp_error(TimeoutError("timed out"))
|
|
self.assertIn("timed out", detail.lower())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|