diff --git a/email_profile/clients/imap/mailbox.py b/email_profile/clients/imap/mailbox.py index 49fee95..4dbe0be 100644 --- a/email_profile/clients/imap/mailbox.py +++ b/email_profile/clients/imap/mailbox.py @@ -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: 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.""" diff --git a/tests/clients/imap/test_mailbox.py b/tests/clients/imap/test_mailbox.py index 3f873bb..1312ce4 100644 --- a/tests/clients/imap/test_mailbox.py +++ b/tests/clients/imap/test_mailbox.py @@ -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 @@ -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") diff --git a/tests/clients/imap/test_write_ops.py b/tests/clients/imap/test_write_ops.py index 4cd67a8..096bc11 100644 --- a/tests/clients/imap/test_write_ops.py +++ b/tests/clients/imap/test_write_ops.py @@ -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 @@ -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):