diff --git a/gateway/platforms/email.py b/gateway/platforms/email.py index a34369263637..b5b5c396647c 100644 --- a/gateway/platforms/email.py +++ b/gateway/platforms/email.py @@ -151,6 +151,24 @@ def _strip_html(html: str) -> str: return text.strip() +def _is_html(text: str) -> bool: + """Return True if text appears to be HTML content.""" + stripped = text.strip() + if stripped.startswith((" MIMEText | MIMEMultipart: + """Return a MIME part for the body — multipart/alternative if HTML detected.""" + if _is_html(body): + alt = MIMEMultipart("alternative") + alt.attach(MIMEText(_strip_html(body), "plain", "utf-8")) + alt.attach(MIMEText(body, "html", "utf-8")) + return alt + return MIMEText(body, "plain", "utf-8") + + def _extract_email_address(raw: str) -> str: """Extract bare email address from 'Name ' format.""" match = re.search(r"<([^>]+)>", raw) @@ -509,7 +527,7 @@ def _send_email( msg_id = f"" msg["Message-ID"] = msg_id - msg.attach(MIMEText(body, "plain", "utf-8")) + msg.attach(_build_body_part(body)) smtp = smtplib.SMTP(self._smtp_host, self._smtp_port, timeout=30) try: @@ -619,7 +637,7 @@ def _send_email_with_attachments( msg["Message-ID"] = msg_id if body: - msg.attach(MIMEText(body, "plain", "utf-8")) + msg.attach(_build_body_part(body)) for file_path in file_paths: p = Path(file_path) @@ -700,7 +718,7 @@ def _send_email_with_attachment( msg["Message-ID"] = msg_id if body: - msg.attach(MIMEText(body, "plain", "utf-8")) + msg.attach(_build_body_part(body)) # Attach file p = Path(file_path) diff --git a/tests/gateway/test_email.py b/tests/gateway/test_email.py index 7c1d0d48e17c..809c89e8678f 100644 --- a/tests/gateway/test_email.py +++ b/tests/gateway/test_email.py @@ -929,6 +929,135 @@ def test_send_email_tool_not_configured(self): self.assertIn("not configured", result["error"]) +class TestHtmlDetection(unittest.TestCase): + """Test _is_html detection and _build_body_part multipart/alternative logic.""" + + def test_plain_text_not_detected_as_html(self): + from gateway.platforms.email import _is_html + self.assertFalse(_is_html("Hello, this is plain text.")) + self.assertFalse(_is_html("No tags here.\nJust lines.")) + self.assertFalse(_is_html("")) + + def test_doctype_detected(self): + from gateway.platforms.email import _is_html + self.assertTrue(_is_html("Hi")) + self.assertTrue(_is_html("\n")) + + def test_html_tag_detected(self): + from gateway.platforms.email import _is_html + self.assertTrue(_is_html("Content")) + self.assertTrue(_is_html("Content")) + + def test_block_level_tags_detected(self): + from gateway.platforms.email import _is_html + self.assertTrue(_is_html("

A paragraph

")) + self.assertTrue(_is_html("
A div
")) + self.assertTrue(_is_html("Hello
World")) + self.assertTrue(_is_html("

Title

")) + self.assertTrue(_is_html("
Heading
")) + self.assertTrue(_is_html("")) + self.assertTrue(_is_html("
  1. Item
")) + self.assertTrue(_is_html("
Cell
")) + + def test_inline_tags_not_detected(self): + from gateway.platforms.email import _is_html + self.assertFalse(_is_html("Use bold text")) + self.assertFalse(_is_html("inline")) + self.assertFalse(_is_html("link")) + + def test_angle_brackets_in_prose_not_detected(self): + from gateway.platforms.email import _is_html + self.assertFalse(_is_html("if x < 10 and y > 5")) + self.assertFalse(_is_html("use as placeholder")) + + def test_leading_whitespace_ignored(self): + from gateway.platforms.email import _is_html + self.assertTrue(_is_html(" \n ")) + self.assertTrue(_is_html("\n\nX")) + + def test_build_body_part_plain_text(self): + from gateway.platforms.email import _build_body_part + part = _build_body_part("Hello, plain text.") + self.assertEqual(part.get_content_type(), "text/plain") + self.assertIn("Hello, plain text.", part.get_payload(decode=True).decode()) + + def test_build_body_part_html_returns_alternative(self): + from gateway.platforms.email import _build_body_part + html = "

Hello

World

" + part = _build_body_part(html) + self.assertEqual(part.get_content_type(), "multipart/alternative") + subparts = part.get_payload() + self.assertEqual(len(subparts), 2) + self.assertEqual(subparts[0].get_content_type(), "text/plain") + self.assertEqual(subparts[1].get_content_type(), "text/html") + # Plain part should have tags stripped + plain_body = subparts[0].get_payload(decode=True).decode() + self.assertNotIn("

", plain_body) + self.assertIn("Hello", plain_body) + # HTML part should preserve original + html_body = subparts[1].get_payload(decode=True).decode() + self.assertIn("

Hello

", html_body) + + def test_build_body_part_with_div_content(self): + from gateway.platforms.email import _build_body_part + html = "
Report:

Revenue is up 20%

" + part = _build_body_part(html) + self.assertEqual(part.get_content_type(), "multipart/alternative") + + def test_send_email_uses_multipart_alternative_for_html(self): + """_send_email should produce multipart/alternative when body is HTML.""" + from gateway.config import PlatformConfig + with patch.dict(os.environ, { + "EMAIL_ADDRESS": "hermes@test.com", + "EMAIL_PASSWORD": "secret", + "EMAIL_IMAP_HOST": "imap.test.com", + "EMAIL_SMTP_HOST": "smtp.test.com", + }): + from gateway.platforms.email import EmailAdapter + adapter = EmailAdapter(PlatformConfig(enabled=True)) + + html_body = "

Report

Details here.

" + + with patch("smtplib.SMTP") as mock_smtp: + mock_server = MagicMock() + mock_smtp.return_value = mock_server + + adapter._send_email("user@test.com", html_body, None) + + sent_msg = mock_server.send_message.call_args[0][0] + # The outer message is multipart/mixed, containing a multipart/alternative + payloads = sent_msg.get_payload() + alt_part = payloads[0] if isinstance(payloads, list) else sent_msg + self.assertEqual(alt_part.get_content_type(), "multipart/alternative") + sub = alt_part.get_payload() + self.assertEqual(sub[0].get_content_type(), "text/plain") + self.assertEqual(sub[1].get_content_type(), "text/html") + + def test_send_email_plain_text_unchanged(self): + """_send_email should send plain MIMEText when body is not HTML.""" + from gateway.config import PlatformConfig + with patch.dict(os.environ, { + "EMAIL_ADDRESS": "hermes@test.com", + "EMAIL_PASSWORD": "secret", + "EMAIL_IMAP_HOST": "imap.test.com", + "EMAIL_SMTP_HOST": "smtp.test.com", + }): + from gateway.platforms.email import EmailAdapter + adapter = EmailAdapter(PlatformConfig(enabled=True)) + + with patch("smtplib.SMTP") as mock_smtp: + mock_server = MagicMock() + mock_smtp.return_value = mock_server + + adapter._send_email("user@test.com", "Just plain text.", None) + + sent_msg = mock_server.send_message.call_args[0][0] + payloads = sent_msg.get_payload() + # Should be a single text/plain part inside the multipart/mixed + text_part = payloads[0] if isinstance(payloads, list) else sent_msg + self.assertEqual(text_part.get_content_type(), "text/plain") + + class TestSmtpConnectionCleanup(unittest.TestCase): """Verify SMTP connections are closed even when send_message raises."""