fix(email): test saved OAuth accounts through shared transports (#5653)

* fix(email): use XOAUTH2 in test-connection for Google OAuth accounts

The test-connection endpoint was password-only and had no awareness of
OAuth accounts. For Google-connected accounts this caused two failures:
- IMAP: "Need IMAP host, username, and password" because imap_pass is
  empty (no password is stored for OAuth accounts)
- SMTP: 535 BadCredentials because smtp.login() was called with an
  empty password instead of an XOAUTH2 token

Fix: include oauth_provider and token fields in saved_body when hydrating
from the DB, then use conn.authenticate("XOAUTH2") / smtp.auth("XOAUTH2")
for Google accounts in both the IMAP and SMTP test paths, mirroring what
_send_smtp_message already does for real sends.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* test(email): add OAuth2 tests for test-connection endpoint

Covers the XOAUTH2 changes made to routes/email_routes.py:
- Google OAuth accounts must not be rejected with 'Need IMAP host,
  username, and password' (no stored password for OAuth accounts)
- IMAP and SMTP test paths must use conn.authenticate('XOAUTH2')
  for Google accounts
- Password accounts must still use conn.login() / smtp.login()

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(email): bind OAuth account tests to Google transport

---------

Co-authored-by: TNTBA <trynottobreakanything@gmail.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

routes/email_helpers.py

@@ -1035,12 +1035,22 @@ def _coerce_imap_timeout_seconds(raw: str | None) -> int:
 _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):
+def _open_imap_connection(
+    host: str,
+    port: int,
+    *,
+    starttls: bool,
+    timeout: int = _IMAP_TIMEOUT_SECONDS,
+    ssl_context=None,
+):
     """Open an IMAP connection using the configured security mode."""
     port = int(port or 993)
     if starttls:
         conn = imaplib.IMAP4(host, port, timeout=timeout)
         try:
+            if ssl_context:
+                conn.starttls(ssl_context=ssl_context)
+            else:
                 conn.starttls()
         except Exception:
             # Don't leak the open plain socket if the STARTTLS upgrade is
@@ -1051,7 +1061,8 @@ def _open_imap_connection(host: str, port: int, *, starttls: bool, timeout: int
                 pass
             raise
     elif port == 993:
-        conn = imaplib.IMAP4_SSL(host, port, timeout=timeout)
+        kwargs = {"ssl_context": ssl_context} if ssl_context else {}
+        conn = imaplib.IMAP4_SSL(host, port, timeout=timeout, **kwargs)
     else:
         conn = imaplib.IMAP4(host, port, timeout=timeout)
     try:

routes/email_routes.py

@@ -20,6 +20,7 @@ import email as email_mod
 import email.header
 import email.utils
 import smtplib
+import ssl
 import json
 import re
 import html
@@ -47,6 +48,7 @@ from routes.email_helpers import (
     _load_settings, _save_settings, _get_email_config,
     _send_smtp_message, _smtp_security_mode,
     _IMAP_TIMEOUT_SECONDS, _open_imap_connection,
+    _get_valid_google_token, _xoauth2_bytes, _xoauth2_raw,
     make_oauth_state, verify_oauth_state,
     EmailNotConfiguredError,
     _imap_connect, _imap, _decode_header, _detect_sent_folder, _detect_drafts_folder,
@@ -65,6 +67,27 @@ logger = logging.getLogger(__name__)
 
 ODYSSEUS_MAIL_ORIGIN = "odysseus-ui"
 EMAIL_READ_ATTACHMENT_VERSION = 2
+_GOOGLE_OAUTH_IMAP_HOST = "imap.gmail.com"
+_GOOGLE_OAUTH_SMTP_HOST = "smtp.gmail.com"
+_SERVER_OWNED_OAUTH_FIELDS = {
+    "oauth_provider",
+    "oauth_access_token",
+    "oauth_refresh_token",
+    "oauth_token_expiry",
+}
+
+
+def _normalized_mail_host(value) -> str:
+    """Normalize a mail hostname for exact provider-bound comparisons."""
+    return str(value or "").strip().lower().rstrip(".")
+
+
+def _google_oauth_imap_transport_allowed(port: int, starttls: bool) -> bool:
+    return (port == 993 and not starttls) or (port == 143 and starttls)
+
+
+def _google_oauth_smtp_transport_allowed(port: int, security: str) -> bool:
+    return (port == 465 and security == "ssl") or (port == 587 and security == "starttls")
 
 
 def _safe_attachment_zip_name(name: str, fallback: str) -> str:
@@ -5013,15 +5036,24 @@ def setup_email_routes():
                     "smtp_security": _smtp_security_mode({"smtp_security": getattr(row, "smtp_security", ""), "smtp_port": row.smtp_port}),
                     "smtp_user": row.smtp_user or "",
                     "smtp_password": _decrypt(row.smtp_password or ""),
+                    "oauth_provider": row.oauth_provider or "",
+                    "oauth_access_token": row.oauth_access_token or "",
+                    "oauth_refresh_token": row.oauth_refresh_token or "",
+                    "oauth_token_expiry": row.oauth_token_expiry or "",
+                    "account_id": acc_id,
                 }
                 for key, value in body.items():
-                    if key == "account_id":
+                    if key == "account_id" or key in _SERVER_OWNED_OAUTH_FIELDS:
                         continue
                     if value not in (None, ""):
                         saved_body[key] = value
                 body = saved_body
             finally:
                 db.close()
+        else:
+            # OAuth state belongs to a saved, owner-checked account. Do not let
+            # inline test payloads select the OAuth branch or supply token data.
+            body = {key: value for key, value in body.items() if key not in _SERVER_OWNED_OAUTH_FIELDS}
 
         imap_result = {"ok": False}
         smtp_result = None
@@ -5031,11 +5063,33 @@ def setup_email_routes():
         imap_user = (body.get("imap_user") or "").strip()
         imap_pass = body.get("imap_password") or ""
         imap_starttls = bool(body.get("imap_starttls"))
+        oauth_provider = body.get("oauth_provider") or ""
+
+        google_token = None
+        google_token_loaded = False
+        google_ssl_context = (
+            ssl.create_default_context()
+            if oauth_provider == "google"
+            else None
+        )
+
+        def _google_token():
+            nonlocal google_token, google_token_loaded
+            if not google_token_loaded:
+                google_token = _get_valid_google_token(body.get("account_id"), body)
+                google_token_loaded = True
+            if not google_token:
+                raise RuntimeError("Google OAuth token unavailable — reconnect the account")
+            return google_token
 
         if imap_port_err:
             imap_result = {"ok": False, "error": imap_port_err}
-        elif not (imap_host and imap_user and imap_pass):
+        elif not (imap_host and imap_user and (imap_pass or oauth_provider == "google")):
             imap_result = {"ok": False, "error": "Need IMAP host, username, and password"}
+        elif oauth_provider == "google" and _normalized_mail_host(imap_host) != _GOOGLE_OAUTH_IMAP_HOST:
+            imap_result = {"ok": False, "error": "Google OAuth IMAP requires imap.gmail.com"}
+        elif oauth_provider == "google" and not _google_oauth_imap_transport_allowed(imap_port, imap_starttls):
+            imap_result = {"ok": False, "error": "Google OAuth IMAP requires TLS on port 993 or STARTTLS on port 143"}
         else:
             # Connection mode resolution:
             #   STARTTLS on  → plain IMAP4 + .starttls() (upgrade)
@@ -5045,13 +5099,22 @@ def setup_email_routes():
             # port (Dovecot on 31143, etc.) would always fail the SSL
             # handshake because they're not actually wrapped in TLS.
             try:
+                imap_kwargs = {
+                    "starttls": imap_starttls,
+                    "timeout": _IMAP_TIMEOUT_SECONDS,
+                }
+                if google_ssl_context:
+                    imap_kwargs["ssl_context"] = google_ssl_context
                 conn = _open_imap_connection(
                     imap_host,
                     imap_port,
-                    starttls=imap_starttls,
+                    **imap_kwargs,
-                    timeout=_IMAP_TIMEOUT_SECONDS,
                 )
                 try:
+                    if oauth_provider == "google":
+                        token = _google_token()
+                        conn.authenticate("XOAUTH2", lambda x: _xoauth2_bytes(imap_user, token))
+                    else:
                         conn.login(imap_user, imap_pass)
                     imap_result = {"ok": True}
                 finally:
@@ -5064,25 +5127,70 @@ def setup_email_routes():
         smtp_port, smtp_port_err = _coerce_port(body.get("smtp_port"), 465)
         if smtp_host and smtp_port_err:
             smtp_result = {"ok": False, "error": smtp_port_err}
+        elif oauth_provider == "google" and smtp_host and _normalized_mail_host(smtp_host) != _GOOGLE_OAUTH_SMTP_HOST:
+            smtp_result = {"ok": False, "error": "Google OAuth SMTP requires smtp.gmail.com"}
+        elif (
+            oauth_provider == "google"
+            and smtp_host
+            and not _google_oauth_smtp_transport_allowed(
+                smtp_port,
+                _smtp_security_mode({"smtp_security": body.get("smtp_security"), "smtp_port": smtp_port}),
+            )
+        ):
+            smtp_result = {"ok": False, "error": "Google OAuth SMTP requires TLS on port 465 or STARTTLS on port 587"}
         elif smtp_host:
             smtp_security = _smtp_security_mode({"smtp_security": body.get("smtp_security"), "smtp_port": smtp_port})
             smtp_user = (body.get("smtp_user") or imap_user).strip()
             smtp_pass = body.get("smtp_password") or imap_pass
+            smtp = None
             try:
                 if smtp_security == "ssl":
-                    smtp = smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=10)
+                    smtp_kwargs = (
+                        {"context": google_ssl_context}
+                        if google_ssl_context
+                        else {}
+                    )
+                    smtp = smtplib.SMTP_SSL(
+                        smtp_host,
+                        smtp_port,
+                        timeout=10,
+                        **smtp_kwargs,
+                    )
                 else:
                     smtp = smtplib.SMTP(smtp_host, smtp_port, timeout=10)
                     if smtp_security == "starttls":
-                        smtp.starttls()
                         try:
+                            if google_ssl_context:
+                                smtp.starttls(context=google_ssl_context)
+                            else:
+                                smtp.starttls()
+                        except Exception:
+                            # STARTTLS failed before the auth cleanup block.
+                            # Close the still-open plaintext socket explicitly.
+                            try:
+                                smtp.close()
+                            except Exception:
+                                pass
+                            smtp = None
+                            raise
+                if oauth_provider == "google":
+                    token = _google_token()
+                    smtp.ehlo()
+                    smtp.auth("XOAUTH2", lambda challenge=None: _xoauth2_raw(smtp_user, token), initial_response_ok=True)
+                else:
                     smtp.login(smtp_user, smtp_pass)
                 smtp_result = {"ok": True}
-                finally:
-                    try: smtp.quit()
-                    except Exception: pass
             except Exception as e:
                 smtp_result = {"ok": False, "error": _friendly_email_auth_error("SMTP", smtp_host, e)}
+            finally:
+                if smtp is not None:
+                    try:
+                        smtp.quit()
+                    except Exception:
+                        try:
+                            smtp.close()
+                        except Exception:
+                            pass
 
         return {
             "ok": imap_result["ok"] and (smtp_result is None or smtp_result["ok"]),

tests/test_email_test_connection_oauth.py

@@ -0,0 +1,436 @@
+"""Tests for Google OAuth2 support in the /api/email/accounts/test endpoint.
+
+Covers the changes made to routes/email_routes.py:
+
+- test_account_config: OAuth accounts must not require a stored password.
+- IMAP and SMTP test paths must use XOAUTH2 for Google accounts.
+- Password accounts must still use conn.login() / smtp.login().
+
+These tests use only in-memory SQLite (via SQLAlchemy) and mock network
+objects — no live email server or real OAuth credentials are needed.
+"""
+
+import time
+import unittest.mock as mock
+
+import pytest
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+def _make_orm_db():
+    """Return (Session, SessionFactory) backed by an isolated in-memory SQLite DB."""
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+    from core.database import Base
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
+    Base.metadata.create_all(engine)
+    Factory = sessionmaker(bind=engine)
+    return Factory(), Factory
+
+
+def _make_orm_account(session, account_id="acct-1", owner="alice", **kwargs):
+    from core.database import EmailAccount
+    row = EmailAccount(
+        id=account_id,
+        owner=owner,
+        name=kwargs.get("name", "Test"),
+        from_address=kwargs.get("from_address", "me@nia.law"),
+        imap_host=kwargs.get("imap_host", "imap.gmail.com"),
+        imap_port=kwargs.get("imap_port", 993),
+        imap_user=kwargs.get("imap_user", "me@nia.law"),
+        imap_starttls=kwargs.get("imap_starttls", False),
+        smtp_host=kwargs.get("smtp_host", "smtp.gmail.com"),
+        smtp_port=kwargs.get("smtp_port", 587),
+        smtp_security=kwargs.get("smtp_security", "starttls"),
+        smtp_user=kwargs.get("smtp_user", "me@nia.law"),
+    )
+    for k, v in kwargs.items():
+        if hasattr(row, k):
+            setattr(row, k, v)
+    session.add(row)
+    session.commit()
+    return row
+
+
+# ── test_connection route: OAuth awareness ────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_test_connection_oauth_account_uses_xoauth2_for_imap_and_smtp():
+    """The saved-account test must use XOAUTH2 for both mail protocols."""
+    from src.secret_storage import encrypt as _enc
+    from routes.email_routes import setup_email_routes
+
+    future_expiry = str(int(time.time()) + 7200)
+    db, Factory = _make_orm_db()
+    _make_orm_account(
+        db, account_id="acct-oauth", owner="alice",
+        oauth_provider="google",
+        oauth_access_token=_enc("ya29.live"),
+        oauth_refresh_token=_enc("1//refresh"),
+        oauth_token_expiry=future_expiry,
+    )
+    db.close()
+
+    router = setup_email_routes()
+    test_conn = None
+    for route in router.routes:
+        if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set()):
+            test_conn = route.endpoint
+            break
+    assert test_conn is not None, "test-connection route not found"
+
+    mock_imap_conn = mock.MagicMock()
+    mock_smtp_conn = mock.MagicMock()
+
+    class _FakeReq:
+        async def json(self):
+            return {"account_id": "acct-oauth"}
+
+    with mock.patch("core.database.SessionLocal", Factory), \
+         mock.patch("routes.email_routes._open_imap_connection", return_value=mock_imap_conn), \
+         mock.patch("routes.email_routes.smtplib.SMTP", return_value=mock_smtp_conn), \
+         mock.patch("routes.email_routes.smtplib.SMTP_SSL", return_value=mock_smtp_conn), \
+         mock.patch("routes.email_routes._get_valid_google_token", return_value="ya29.live") as token_getter:
+        result = await test_conn(req=_FakeReq(), owner="alice")
+
+    assert result["ok"] is True
+    assert result["imap"].get("ok") is True, \
+        f"OAuth IMAP test must succeed, got: {result['imap']}"
+    assert result["smtp"].get("ok") is True, \
+        f"OAuth SMTP test must succeed, got: {result['smtp']}"
+    mock_imap_conn.authenticate.assert_called_once()
+    assert mock_imap_conn.authenticate.call_args[0][0] == "XOAUTH2"
+    mock_imap_conn.login.assert_not_called()
+    mock_smtp_conn.auth.assert_called_once()
+    assert mock_smtp_conn.auth.call_args[0][0] == "XOAUTH2"
+    mock_smtp_conn.login.assert_not_called()
+    token_getter.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_test_connection_password_account_still_uses_login():
+    """Existing password accounts must still go through the login() path."""
+    from src.secret_storage import encrypt as _enc
+    from routes.email_routes import setup_email_routes
+
+    db, Factory = _make_orm_db()
+    _make_orm_account(
+        db, account_id="acct-pw", owner="alice",
+        imap_host="imap.example.com",
+        imap_user="me@example.com",
+        smtp_host="smtp.example.com",
+        smtp_user="me@example.com",
+        imap_password=_enc("hunter2"),
+    )
+    db.close()
+
+    router = setup_email_routes()
+    test_conn = None
+    for route in router.routes:
+        if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set()):
+            test_conn = route.endpoint
+            break
+
+    mock_imap_conn = mock.MagicMock()
+    mock_smtp_conn = mock.MagicMock()
+
+    class _FakeReq:
+        async def json(self):
+            return {"account_id": "acct-pw"}
+
+    with mock.patch("core.database.SessionLocal", Factory), \
+         mock.patch("routes.email_routes._open_imap_connection", return_value=mock_imap_conn), \
+         mock.patch("routes.email_routes.smtplib.SMTP", return_value=mock_smtp_conn), \
+         mock.patch("routes.email_routes.smtplib.SMTP_SSL", return_value=mock_smtp_conn):
+        result = await test_conn(req=_FakeReq(), owner="alice")
+
+    assert result["ok"] is True
+    mock_imap_conn.login.assert_called_once_with("me@example.com", "hunter2")
+    mock_imap_conn.authenticate.assert_not_called()
+    mock_smtp_conn.login.assert_called_once_with("me@example.com", "hunter2")
+    mock_smtp_conn.auth.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_test_connection_rejects_non_google_hosts_before_oauth_auth():
+    """Saved Google tokens must never be routed to edited custom hosts."""
+    from src.secret_storage import encrypt as _enc
+    from routes.email_routes import setup_email_routes
+
+    db, Factory = _make_orm_db()
+    _make_orm_account(
+        db,
+        account_id="acct-oauth",
+        owner="alice",
+        oauth_provider="google",
+        oauth_access_token=_enc("ya29.live"),
+        oauth_refresh_token=_enc("1//refresh"),
+        oauth_token_expiry=str(int(time.time()) + 7200),
+    )
+    db.close()
+
+    router = setup_email_routes()
+    test_conn = next(
+        route.endpoint
+        for route in router.routes
+        if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set())
+    )
+
+    class _FakeReq:
+        async def json(self):
+            return {
+                "account_id": "acct-oauth",
+                "imap_host": "collector.invalid",
+                "smtp_host": "collector.invalid",
+            }
+
+    with mock.patch("core.database.SessionLocal", Factory), \
+         mock.patch("routes.email_routes._open_imap_connection") as open_imap, \
+         mock.patch("routes.email_routes.smtplib.SMTP") as open_smtp, \
+         mock.patch("routes.email_routes.smtplib.SMTP_SSL") as open_smtp_ssl, \
+         mock.patch("routes.email_routes._get_valid_google_token") as token_getter:
+        result = await test_conn(req=_FakeReq(), owner="alice")
+
+    assert result["ok"] is False
+    assert "imap.gmail.com" in result["imap"]["error"]
+    assert "smtp.gmail.com" in result["smtp"]["error"]
+    open_imap.assert_not_called()
+    open_smtp.assert_not_called()
+    open_smtp_ssl.assert_not_called()
+    token_getter.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_test_connection_rejects_insecure_oauth_transports_before_auth():
+    """Google OAuth credentials must not be tested over plaintext transports."""
+    from src.secret_storage import encrypt as _enc
+    from routes.email_routes import setup_email_routes
+
+    db, Factory = _make_orm_db()
+    _make_orm_account(
+        db,
+        account_id="acct-oauth",
+        owner="alice",
+        oauth_provider="google",
+        oauth_access_token=_enc("ya29.live"),
+        oauth_refresh_token=_enc("1//refresh"),
+        oauth_token_expiry=str(int(time.time()) + 7200),
+    )
+    db.close()
+
+    router = setup_email_routes()
+    test_conn = next(
+        route.endpoint
+        for route in router.routes
+        if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set())
+    )
+
+    class _FakeReq:
+        async def json(self):
+            return {
+                "account_id": "acct-oauth",
+                "imap_port": 143,
+                "imap_starttls": False,
+                "smtp_port": 587,
+                "smtp_security": "none",
+            }
+
+    with mock.patch("core.database.SessionLocal", Factory), \
+         mock.patch("routes.email_routes._open_imap_connection") as open_imap, \
+         mock.patch("routes.email_routes.smtplib.SMTP") as open_smtp, \
+         mock.patch("routes.email_routes.smtplib.SMTP_SSL") as open_smtp_ssl, \
+         mock.patch("routes.email_routes._get_valid_google_token") as token_getter:
+        result = await test_conn(req=_FakeReq(), owner="alice")
+
+    assert result["ok"] is False
+    assert "TLS" in result["imap"]["error"]
+    assert "TLS" in result["smtp"]["error"]
+    open_imap.assert_not_called()
+    open_smtp.assert_not_called()
+    open_smtp_ssl.assert_not_called()
+    token_getter.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_test_connection_does_not_accept_inline_oauth_state():
+    """Only an owner-checked saved account may select the OAuth branch."""
+    from routes.email_routes import setup_email_routes
+
+    router = setup_email_routes()
+    test_conn = next(
+        route.endpoint
+        for route in router.routes
+        if route.path == "/api/email/accounts/test" and "POST" in getattr(route, "methods", set())
+    )
+
+    class _FakeReq:
+        async def json(self):
+            return {
+                "imap_host": "imap.gmail.com",
+                "imap_user": "me@example.com",
+                "oauth_provider": "google",
+                "oauth_access_token": "client-supplied-token",
+            }
+
+    with mock.patch("routes.email_routes._open_imap_connection") as open_imap, \
+         mock.patch("routes.email_routes._get_valid_google_token") as token_getter:
+        result = await test_conn(req=_FakeReq(), owner="alice")
+
+    assert result["ok"] is False
+    assert result["imap"]["error"] == "Need IMAP host, username, and password"
+    open_imap.assert_not_called()
+    token_getter.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("imap_starttls", "imap_port"),
+    [(False, 993), (True, 143)],
+)
+async def test_test_connection_verifies_imap_tls_before_loading_oauth_token(
+    imap_starttls,
+    imap_port,
+):
+    """Both Google IMAP TLS modes receive a certificate-verifying context;
+    certificate rejection happens before the bearer token is loaded."""
+    import ssl
+
+    from routes.email_routes import setup_email_routes
+    from src.secret_storage import encrypt as _enc
+
+    db, Factory = _make_orm_db()
+    _make_orm_account(
+        db,
+        account_id="acct-imap-tls",
+        owner="alice",
+        imap_port=imap_port,
+        imap_starttls=imap_starttls,
+        smtp_host="",
+        oauth_provider="google",
+        oauth_access_token=_enc("ya29.live"),
+        oauth_refresh_token=_enc("1//refresh"),
+        oauth_token_expiry=str(int(time.time()) + 7200),
+    )
+    db.close()
+
+    router = setup_email_routes()
+    test_conn = next(
+        route.endpoint
+        for route in router.routes
+        if route.path == "/api/email/accounts/test"
+        and "POST" in getattr(route, "methods", set())
+    )
+
+    class _FakeReq:
+        async def json(self):
+            return {"account_id": "acct-imap-tls"}
+
+    context = ssl.create_default_context()
+    starttls_conn = mock.MagicMock()
+    starttls_conn.starttls.side_effect = ssl.SSLCertVerificationError(
+        "untrusted certificate"
+    )
+    with mock.patch("core.database.SessionLocal", Factory), \
+         mock.patch(
+             "routes.email_routes.ssl.create_default_context",
+             return_value=context,
+         ), mock.patch(
+             "routes.email_helpers.imaplib.IMAP4",
+             return_value=starttls_conn,
+         ) as imap_cls, mock.patch(
+             "routes.email_helpers.imaplib.IMAP4_SSL",
+             side_effect=ssl.SSLCertVerificationError("untrusted certificate"),
+         ) as imap_ssl_cls, mock.patch(
+             "routes.email_routes._get_valid_google_token"
+         ) as token_getter:
+        result = await test_conn(req=_FakeReq(), owner="alice")
+
+    assert result["ok"] is False
+    assert result["imap"]["ok"] is False
+    assert context.check_hostname is True
+    assert context.verify_mode == ssl.CERT_REQUIRED
+    token_getter.assert_not_called()
+    if imap_starttls:
+        assert starttls_conn.starttls.call_args.kwargs["ssl_context"] is context
+        imap_ssl_cls.assert_not_called()
+    else:
+        assert imap_ssl_cls.call_args.kwargs["ssl_context"] is context
+        imap_cls.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("smtp_security", "smtp_port"),
+    [("ssl", 465), ("starttls", 587)],
+)
+async def test_test_connection_verifies_smtp_tls_before_loading_oauth_token(
+    smtp_security,
+    smtp_port,
+):
+    """Both Google SMTP TLS modes reject an invalid certificate before any
+    XOAUTH2 credential is obtained or sent."""
+    import ssl
+
+    from routes.email_routes import setup_email_routes
+    from src.secret_storage import encrypt as _enc
+
+    db, Factory = _make_orm_db()
+    _make_orm_account(
+        db,
+        account_id="acct-smtp-tls",
+        owner="alice",
+        imap_host="",
+        smtp_port=smtp_port,
+        smtp_security=smtp_security,
+        oauth_provider="google",
+        oauth_access_token=_enc("ya29.live"),
+        oauth_refresh_token=_enc("1//refresh"),
+        oauth_token_expiry=str(int(time.time()) + 7200),
+    )
+    db.close()
+
+    router = setup_email_routes()
+    test_conn = next(
+        route.endpoint
+        for route in router.routes
+        if route.path == "/api/email/accounts/test"
+        and "POST" in getattr(route, "methods", set())
+    )
+
+    class _FakeReq:
+        async def json(self):
+            return {"account_id": "acct-smtp-tls"}
+
+    context = ssl.create_default_context()
+    starttls_smtp = mock.MagicMock()
+    starttls_smtp.starttls.side_effect = ssl.SSLCertVerificationError(
+        "untrusted certificate"
+    )
+    with mock.patch("core.database.SessionLocal", Factory), \
+         mock.patch(
+             "routes.email_routes.ssl.create_default_context",
+             return_value=context,
+         ), mock.patch(
+             "routes.email_routes.smtplib.SMTP",
+             return_value=starttls_smtp,
+         ) as smtp_cls, mock.patch(
+             "routes.email_routes.smtplib.SMTP_SSL",
+             side_effect=ssl.SSLCertVerificationError("untrusted certificate"),
+         ) as smtp_ssl_cls, mock.patch(
+             "routes.email_routes._get_valid_google_token"
+         ) as token_getter:
+        result = await test_conn(req=_FakeReq(), owner="alice")
+
+    assert result["ok"] is False
+    assert result["smtp"]["ok"] is False
+    assert context.check_hostname is True
+    assert context.verify_mode == ssl.CERT_REQUIRED
+    token_getter.assert_not_called()
+    if smtp_security == "starttls":
+        assert starttls_smtp.starttls.call_args.kwargs["context"] is context
+        assert starttls_smtp.close.called
+        smtp_ssl_cls.assert_not_called()
+    else:
+        assert smtp_ssl_cls.call_args.kwargs["context"] is context
+        smtp_cls.assert_not_called()