Use shared IMAP timeout for account tests (#1088)

routes/email_helpers.py

@@ -608,7 +608,32 @@ def _list_email_accounts() -> list[dict]:
 
 # ── IMAP helpers ──
 
-_IMAP_TIMEOUT_SECONDS = 15
+def _coerce_imap_timeout_seconds(raw: str | None) -> int:
+    try:
+        value = int(raw or "30")
+    except (TypeError, ValueError):
+        value = 30
+    return max(5, min(value, 300))
+
+
+_IMAP_TIMEOUT_SECONDS = _coerce_imap_timeout_seconds(os.environ.get("ODYSSEUS_IMAP_TIMEOUT_SECONDS"))
+
+
+def _open_imap_connection(host: str, port: int, *, starttls: bool, timeout: int = _IMAP_TIMEOUT_SECONDS):
+    """Open an IMAP connection using the configured security mode."""
+    port = int(port or 993)
+    if starttls:
+        conn = imaplib.IMAP4(host, port, timeout=timeout)
+        conn.starttls()
+    elif port == 993:
+        conn = imaplib.IMAP4_SSL(host, port, timeout=timeout)
+    else:
+        conn = imaplib.IMAP4(host, port, timeout=timeout)
+    try:
+        conn.sock.settimeout(timeout)
+    except Exception:
+        pass
+    return conn
 
 def _imap_connect(account_id: str | None = None, owner: str = ""):
     # SECURITY: passing `owner` scopes the fallback config lookup so a brand
@@ -622,17 +647,12 @@ def _imap_connect(account_id: str | None = None, owner: str = ""):
     # The last branch is critical: previously this fell into IMAP4_SSL
     # for any non-STARTTLS port, which would fail the TLS handshake on
     # plain local servers (Dovecot on 31143, etc.).
-    if cfg.get("imap_starttls"):
-        conn = imaplib.IMAP4(cfg["imap_host"], cfg["imap_port"], timeout=_IMAP_TIMEOUT_SECONDS)
-        conn.starttls()
-    elif int(cfg.get("imap_port") or 993) == 993:
-        conn = imaplib.IMAP4_SSL(cfg["imap_host"], cfg["imap_port"], timeout=_IMAP_TIMEOUT_SECONDS)
-    else:
-        conn = imaplib.IMAP4(cfg["imap_host"], cfg["imap_port"], timeout=_IMAP_TIMEOUT_SECONDS)
-    try:
-        conn.sock.settimeout(_IMAP_TIMEOUT_SECONDS)
-    except Exception:
-        pass
+    conn = _open_imap_connection(
+        cfg["imap_host"],
+        cfg["imap_port"],
+        starttls=bool(cfg.get("imap_starttls")),
+        timeout=_IMAP_TIMEOUT_SECONDS,
+    )
     conn.login(cfg["imap_user"], cfg["imap_password"])
     return conn
 

routes/email_routes.py

@@ -17,7 +17,6 @@ import sqlite3 as _sql3
 import email as email_mod
 import email.header
 import email.utils
-import imaplib
 import smtplib
 import json
 import re
@@ -41,6 +40,7 @@ from routes.email_helpers import (
     _q, _attach_compose_uploads, _cleanup_compose_uploads,
     _load_settings, _save_settings, _get_email_config,
     _send_smtp_message, _smtp_security_mode,
+    _IMAP_TIMEOUT_SECONDS, _open_imap_connection,
     _imap_connect, _imap, _decode_header, _detect_sent_folder, _detect_drafts_folder,
     _extract_attachment_text, _list_attachments_from_msg,
     _extract_attachment_to_disk, _extract_html, _extract_text,
@@ -3103,13 +3103,12 @@ def setup_email_routes():
             # port (Dovecot on 31143, etc.) would always fail the SSL
             # handshake because they're not actually wrapped in TLS.
             try:
-                if imap_starttls:
-                    conn = imaplib.IMAP4(imap_host, imap_port, timeout=10)
-                    conn.starttls()
-                elif imap_port == 993:
-                    conn = imaplib.IMAP4_SSL(imap_host, imap_port, timeout=10)
-                else:
-                    conn = imaplib.IMAP4(imap_host, imap_port, timeout=10)
+                conn = _open_imap_connection(
+                    imap_host,
+                    imap_port,
+                    starttls=imap_starttls,
+                    timeout=_IMAP_TIMEOUT_SECONDS,
+                )
                 try:
                     conn.login(imap_user, imap_pass)
                     imap_result = {"ok": True}

tests/test_email_imap_timeout.py

@@ -0,0 +1,126 @@
+import os
+import tempfile
+from pathlib import Path
+
+import pytest
+
+_tmp_data = Path(tempfile.mkdtemp(prefix="odysseus-email-imap-test-"))
+os.environ.setdefault("DATA_DIR", str(_tmp_data))
+os.environ.setdefault("DATABASE_URL", f"sqlite:///{_tmp_data / 'app.db'}")
+
+from routes.email_helpers import (
+    _IMAP_TIMEOUT_SECONDS,
+    _coerce_imap_timeout_seconds,
+    _open_imap_connection,
+)
+
+
+class _FakeSock:
+    def __init__(self):
+        self.timeout = None
+
+    def settimeout(self, timeout):
+        self.timeout = timeout
+
+
+class _FakeIMAP:
+    calls = []
+
+    def __init__(self, host, port, timeout=None):
+        self.host = host
+        self.port = port
+        self.timeout = timeout
+        self.sock = _FakeSock()
+        self.starttls_called = False
+        _FakeIMAP.calls.append(("connect", self.__class__.__name__, host, port, timeout))
+
+    def starttls(self):
+        self.starttls_called = True
+        _FakeIMAP.calls.append(("starttls", self.host, self.port))
+
+    def login(self, user, password):
+        _FakeIMAP.calls.append(("login", user, password))
+
+    def logout(self):
+        _FakeIMAP.calls.append(("logout", self.host, self.port))
+
+
+class _FakeIMAPSSL(_FakeIMAP):
+    pass
+
+
+def test_imap_timeout_defaults_and_clamps():
+    assert _coerce_imap_timeout_seconds(None) == 30
+    assert _coerce_imap_timeout_seconds("nonsense") == 30
+    assert _coerce_imap_timeout_seconds("2") == 5
+    assert _coerce_imap_timeout_seconds("999") == 300
+
+
+def test_open_imap_connection_uses_shared_timeout_for_implicit_ssl(monkeypatch):
+    import routes.email_helpers as helpers
+
+    _FakeIMAP.calls = []
+    monkeypatch.setattr(helpers.imaplib, "IMAP4", _FakeIMAP)
+    monkeypatch.setattr(helpers.imaplib, "IMAP4_SSL", _FakeIMAPSSL)
+
+    conn = _open_imap_connection("imap.one.com", 993, starttls=False)
+
+    assert _FakeIMAP.calls == [
+        ("connect", "_FakeIMAPSSL", "imap.one.com", 993, _IMAP_TIMEOUT_SECONDS)
+    ]
+    assert conn.sock.timeout == _IMAP_TIMEOUT_SECONDS
+
+
+def test_open_imap_connection_supports_starttls(monkeypatch):
+    import routes.email_helpers as helpers
+
+    _FakeIMAP.calls = []
+    monkeypatch.setattr(helpers.imaplib, "IMAP4", _FakeIMAP)
+    monkeypatch.setattr(helpers.imaplib, "IMAP4_SSL", _FakeIMAPSSL)
+
+    _open_imap_connection("imap.local", 143, starttls=True)
+
+    assert _FakeIMAP.calls == [
+        ("connect", "_FakeIMAP", "imap.local", 143, _IMAP_TIMEOUT_SECONDS),
+        ("starttls", "imap.local", 143),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_account_config_uses_shared_imap_timeout(monkeypatch):
+    import routes.email_routes as email_routes
+
+    captured = {}
+
+    class _Conn:
+        def login(self, user, password):
+            captured["login"] = (user, password)
+
+        def logout(self):
+            captured["logout"] = True
+
+    def fake_open(host, port, *, starttls, timeout):
+        captured["open"] = (host, port, starttls, timeout)
+        return _Conn()
+
+    class _Req:
+        async def json(self):
+            return {
+                "imap_host": "imap.one.com",
+                "imap_port": 993,
+                "imap_user": "user@example.com",
+                "imap_password": "pw",
+                "imap_starttls": False,
+            }
+
+    monkeypatch.setattr(email_routes, "_open_imap_connection", fake_open)
+
+    router = email_routes.setup_email_routes()
+    endpoint = next(route.endpoint for route in router.routes if route.path == "/api/email/accounts/test")
+
+    result = await endpoint(_Req(), owner="")
+
+    assert result["imap"] == {"ok": True}
+    assert captured["open"] == ("imap.one.com", 993, False, _IMAP_TIMEOUT_SECONDS)
+    assert captured["login"] == ("user@example.com", "pw")
+    assert captured["logout"] is True