refactor(email): fold #92979 review findings into the IMAP ID gate

Two shape fixes from the review of #92979, no behavior change:

- Read imap.capabilities directly instead of the getattr(...) or ()
  fallback — a real imaplib.IMAP4 always sets the attribute in
  _connect() (raising if the server sends no CAPABILITY), so the
  default branch only existed for a MagicMock(spec=["xatom"]) that no
  production path produces. Drop the test that pinned it.
- Trim tests to the invariant bar: the four mock tests duplicated what
  test_email_imap_id_protocol.py proves on the wire (real imaplib, real
  socket, command order), and the phase parametrize (startup vs poll)
  exercised the same single _send_imap_id call site twice. Keep the
  three id_mode cases through the poll path.

42 tests pass; mutation check: guard removed -> the absent-ID case
fails with the #39856 SELECT error, restored -> green.
This commit is contained in:
kshitijk4poor
2026-09-06 21:29:58 +05:30
committed by kshitij
parent 8fd3e08ab9
commit 66f1668850
3 changed files with 4 additions and 67 deletions
+1 -2
View File
@@ -148,8 +148,7 @@ def _send_imap_id(imap: "imaplib.IMAP4") -> None:
connection, which imaplib cannot surface here — the failure appears one command later
as a misleading SELECT error and the adapter retries forever (Purelymail, #39856).
``imap.capabilities`` is populated by imaplib at connect, so the check is free."""
caps = getattr(imap, "capabilities", ()) or ()
if "ID" not in caps:
if "ID" not in imap.capabilities:
logger.debug(
"[Email] Server does not advertise IMAP ID capability; skipping ID"
)
-50
View File
@@ -963,56 +963,6 @@ class TestImapIdExtensionForNetEase(unittest.TestCase):
self.assertIn("login", names)
self.assertLess(names.index("login"), names.index("xatom"))
def test_send_imap_id_sent_when_capability_advertised(self):
"""ID goes out when the server lists it in CAPABILITY."""
from plugins.platforms.email.adapter import _send_imap_id
mock_imap = MagicMock()
mock_imap.capabilities = ("IMAP4REV1", "ID", "UIDPLUS")
_send_imap_id(mock_imap)
mock_imap.xatom.assert_called_once()
self.assertEqual(mock_imap.xatom.call_args.args[0], "ID")
def test_send_imap_id_skipped_when_capability_absent(self):
"""Servers that do not advertise ID must not receive it: Purelymail
answers the unknown command with an untagged ``* BYE Unknown
command.`` and drops the connection, which imaplib surfaces one
command later as a misleading SELECT failure."""
from plugins.platforms.email.adapter import _send_imap_id
mock_imap = MagicMock()
mock_imap.capabilities = ("IMAP4REV1", "UIDPLUS")
_send_imap_id(mock_imap)
mock_imap.xatom.assert_not_called()
def test_send_imap_id_skipped_when_capabilities_attribute_missing(self):
"""A connection object without a capabilities attribute must not
crash the helper; fail toward not sending optional commands."""
from plugins.platforms.email.adapter import _send_imap_id
mock_imap = MagicMock(spec=["xatom"])
_send_imap_id(mock_imap)
mock_imap.xatom.assert_not_called()
def test_send_imap_id_rejection_still_swallowed(self):
"""A server that advertises ID but rejects it with a tagged error
keeps the existing best-effort handling: no exception escapes."""
from plugins.platforms.email.adapter import _send_imap_id
mock_imap = MagicMock()
mock_imap.capabilities = ("IMAP4REV1", "ID")
mock_imap.xatom.side_effect = Exception("BAD ID rejected")
_send_imap_id(mock_imap) # must not raise
mock_imap.xatom.assert_called_once()
class TestConnectSmtp(unittest.TestCase):
"""Test _connect_smtp() helper: protocol selection and IPv6 fallback."""
+3 -15
View File
@@ -4,7 +4,6 @@ No external mail service or credentials are used. Unsupported ID gets BAD
followed by BYE: swallowing the ID error still leaves SELECT unable to proceed.
"""
import asyncio
import imaplib
import socketserver
import threading
@@ -80,8 +79,7 @@ def imap_peer(id_mode):
@pytest.mark.parametrize("id_mode", ["absent", "accept", "reject"])
@pytest.mark.parametrize("phase", ["startup", "poll"])
def test_id_negotiation_preserves_inbox_connection(monkeypatch, id_mode, phase):
def test_id_negotiation_preserves_inbox_connection(monkeypatch, id_mode):
from gateway.config import PlatformConfig
from plugins.platforms.email.adapter import EmailAdapter
@@ -104,18 +102,8 @@ def test_id_negotiation_preserves_inbox_connection(monkeypatch, id_mode, phase):
lambda *args, **kwargs: imaplib.IMAP4(*address, timeout=5),
)
if phase == "startup":
async def connect_and_stop():
try:
return await adapter.connect()
finally:
await adapter.disconnect()
assert asyncio.run(connect_and_stop()) is True
else:
assert adapter._fetch_new_messages() == []
assert adapter._last_fetch_failed is False
assert adapter._fetch_new_messages() == []
assert adapter._last_fetch_failed is False
assert commands.count("ID") == (0 if id_mode == "absent" else 1)
assert commands.index("LOGIN") < commands.index("SELECT") < commands.index("UID")