Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions email_profile/clients/imap/mailbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -184,17 +184,28 @@ def copy(self, target: UIDLike, destination: str) -> None:
def move(self, target: UIDLike, destination: str) -> None:
"""Move a message into another mailbox.

Prefers ``UID MOVE`` (RFC 6851). Falls back to copy + delete + expunge
when the server does not advertise the MOVE extension.
Prefers ``UID MOVE`` (RFC 6851). Falls back to copy + delete +
``UID EXPUNGE`` (RFC 4315) when the server does not advertise MOVE.
``imaplib.IMAP4.error`` and ``IMAPError`` (raised by ``Status.state``
on tagged ``NO``/``BAD`` responses) are treated as the missing-MOVE
signal — network/auth/quota errors propagate.
"""
from email_profile.core.errors import IMAPError

Status.state(self._client.select(_quote(self.name)))
uid = _uid_of(target)
try:

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Blocking]

Problem — narrowing the except from Exception to imaplib.IMAP4.error skips the most common way servers signal "MOVE not supported": a tagged NO / BAD response. Status.state(...) translates those into email_profile.core.errors.IMAPError, which is not a subclass of imaplib.IMAP4.error. So when the server replies NO [CANNOT] MOVE not supported, IMAPError propagates out of move() and the COPY+STORE+UID EXPUNGE fallback never runs — a regression vs. the previous bare except Exception.

Failure scenario — Dovecot (and many others) advertise capabilities through CAPABILITY, not through raising imaplib protocol errors. Calling mailbox.move(uid, "Archive") on a server without RFC 6851 will:

  1. client.uid("MOVE", ...) returns ("NO", [b"MOVE not supported"])
  2. Status.state raises IMAPError("Command failed")
  3. except imaplib.IMAP4.error does not match → caller sees IMAPError, message not moved.

The existing test test_falls_back_to_copy_uid_expunge masks this because it mocks side_effect=imaplib.IMAP4.error(...) — but Status.state would never raise that class in production.

Fix — catch both exception classes (or, better, check capabilities up-front and skip the try entirely when MOVE is unsupported):

from email_profile.core.errors import IMAPError

try:
    Status.state(self._client.uid("MOVE", uid, destination))
except (imaplib.IMAP4.error, IMAPError):
    Status.state(self._client.uid("COPY", uid, destination))
    Status.state(self._client.uid("STORE", uid, "+FLAGS", "\Deleted"))
    try:
        Status.state(self._client.uid("EXPUNGE", uid))
    except (imaplib.IMAP4.error, IMAPError) as exc:
        raise IMAPError(
            "Server lacks MOVE and UIDPLUS; message was copied and "
            "flagged \Deleted but not expunged."
        ) from exc

Add a regression test that simulates ("NO", [b"MOVE not supported"]) returned from uid() (no exception raised) and asserts the COPY/STORE/UID-EXPUNGE path runs.

References — RFC 3501 §6.4 (untagged NO for unsupported commands), RFC 6851 §3 (MOVE capability advertised via CAPABILITY; servers without it return NO/BAD).

Status.state(self._client.uid("MOVE", uid, destination))
except Exception:
except (imaplib.IMAP4.error, IMAPError):
Status.state(self._client.uid("COPY", uid, destination))
Status.state(self._client.uid("STORE", uid, "+FLAGS", "\\Deleted"))
Status.state(self._client.expunge())
try:
Status.state(self._client.uid("EXPUNGE", uid))
except (imaplib.IMAP4.error, IMAPError) as exc:
raise IMAPError(
"Server lacks MOVE and UIDPLUS; message was copied and "
"flagged \\Deleted but not expunged."
) from exc

def create(self) -> None:
"""Create this mailbox on the server."""
Expand Down
91 changes: 91 additions & 0 deletions tests/clients/imap/test_mailbox.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import imaplib
from unittest import TestCase
from unittest.mock import patch

from email_profile import Email, Message, Q, Query
from email_profile.core.errors import IMAPError
from tests.conftest import SAMPLE_RFC822, make_fake_client


Expand Down Expand Up @@ -81,3 +83,92 @@ def test_returns_appended_uid_when_supported(self):
def test_returns_none_without_uidplus(self):
self.fake.append.return_value = ("OK", [b"APPEND completed"])
self.assertIsNone(self.app.inbox.append(SAMPLE_RFC822))


class TestMailBoxMove(TestCase):
def setUp(self):
self.fake = make_fake_client()
self._patcher = patch(
"email_profile.clients.imap.client.imaplib.IMAP4_SSL",
return_value=self.fake,
)
self._patcher.start()
self.app = Email("imap.x", "u", "p").connect()

def tearDown(self):
self.app.close()
self._patcher.stop()

def test_uses_move_when_supported(self):
self.app.inbox.move("42", "Archive")
calls = [c.args[0] for c in self.fake.uid.call_args_list]
self.assertIn("MOVE", calls)
self.assertNotIn("COPY", calls)
self.fake.expunge.assert_not_called()

def test_falls_back_to_copy_uid_expunge(self):
def side(command, *args):
cmd = command.upper()
if cmd == "MOVE":
raise imaplib.IMAP4.error("MOVE not supported")
return ("OK", [b"Done"])

self.fake.uid.side_effect = side
self.app.inbox.move("42", "Archive")
calls = [c.args[0] for c in self.fake.uid.call_args_list]
self.assertIn("COPY", calls)
self.assertIn("STORE", calls)
self.assertIn("EXPUNGE", calls)
self.fake.expunge.assert_not_called()

def test_propagates_non_imap_errors(self):
self.fake.uid.side_effect = ConnectionResetError("network drop")
with self.assertRaises(ConnectionResetError):
self.app.inbox.move("42", "Archive")

def test_raises_when_uidplus_missing(self):
def side(command, *args):
cmd = command.upper()
if cmd in {"MOVE", "EXPUNGE"}:
raise imaplib.IMAP4.error("not supported")
return ("OK", [b"Done"])

self.fake.uid.side_effect = side
with self.assertRaises(IMAPError):
self.app.inbox.move("42", "Archive")

def test_falls_back_when_server_responds_no_to_move(self):
"""Server signals missing MOVE via tagged ``NO`` — not an exception.

``Status.state`` translates ``("NO", ...)`` into ``IMAPError``;
the fallback must catch it so the COPY+STORE+UID EXPUNGE path runs.
"""

def side(command, *args):
cmd = command.upper()
if cmd == "MOVE":
return ("NO", [b"MOVE not supported"])
return ("OK", [b"Done"])

self.fake.uid.side_effect = side
self.app.inbox.move("42", "Archive")
calls = [c.args[0] for c in self.fake.uid.call_args_list]
self.assertIn("COPY", calls)
self.assertIn("STORE", calls)
self.assertIn("EXPUNGE", calls)
self.fake.expunge.assert_not_called()

def test_raises_when_server_responds_no_to_uid_expunge(self):
"""Server lacks UIDPLUS and replies ``NO`` to UID EXPUNGE."""

def side(command, *args):
cmd = command.upper()
if cmd == "MOVE":
return ("NO", [b"MOVE not supported"])
if cmd == "EXPUNGE":
return ("NO", [b"UIDPLUS not supported"])
return ("OK", [b"Done"])

self.fake.uid.side_effect = side
with self.assertRaises(IMAPError):
self.app.inbox.move("42", "Archive")
7 changes: 5 additions & 2 deletions tests/clients/imap/test_write_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,12 +79,14 @@ def test_move_prefers_uid_move(self):
self.assertEqual(call.args, ("MOVE", "42", "Archive"))

def test_move_falls_back_to_copy_delete(self):
import imaplib

calls = []

def fake_uid(command, *args):
calls.append((command, args))
if command.upper() == "MOVE":
raise Exception("MOVE not supported")
raise imaplib.IMAP4.error("MOVE not supported")
return ("OK", [b"Done"])

self.fake.uid.side_effect = fake_uid
Expand All @@ -94,7 +96,8 @@ def fake_uid(command, *args):
self.assertIn("MOVE", commands)
self.assertIn("COPY", commands)
self.assertIn("STORE", commands)
self.fake.expunge.assert_called_once()
self.assertIn("EXPUNGE", commands)
self.fake.expunge.assert_not_called()


class TestMailboxAdmin(_WriteTest):
Expand Down
Loading