amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-03-24 23:42:45 +00:00
Code Issues Projects Releases Wiki Activity

Add IMAP move/delete compatibility fallbacks (#671)

Browse Source 
* Add IMAP move/delete compatibility fallbacks with tests

* Expand IMAP fallback tests for success and error paths
This commit is contained in:
Kili
2026-03-09 22:29:01 +01:00
committed by GitHub
parent c4d7455839
commit 6e6c90e19b
2 changed files with 76 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
parsedmarc/mail/imap.py
View File

@@ -55,10 +55,28 @@ class IMAPConnection(MailboxConnection):
return cast(str, self._client.fetch_message(message_id, parse=False))
def delete_message(self, message_id: int):
self._client.delete_messages([message_id])
try:
self._client.delete_messages([message_id])
except IMAPClientError as error:
logger.warning(
"IMAP delete fallback for message %s due to server error: %s",
message_id,
error,
)
self._client.add_flags([message_id], [r"\Deleted"], silent=True)
self._client.expunge()
def move_message(self, message_id: int, folder_name: str):
self._client.move_messages([message_id], folder_name)
try:
self._client.move_messages([message_id], folder_name)
except IMAPClientError as error:
logger.warning(
"IMAP move fallback for message %s due to server error: %s",
message_id,
error,
)
self._client.copy([message_id], folder_name)
self.delete_message(message_id)
def keepalive(self):
self._client.noop()

58
tests.py
View File

@@ -278,8 +278,6 @@ aws_service = aoss
self.assertEqual(mock_set_hosts.call_args.kwargs.get("auth_type"), "awssigv4")
self.assertEqual(mock_set_hosts.call_args.kwargs.get("aws_region"), "eu-west-1")
self.assertEqual(mock_set_hosts.call_args.kwargs.get("aws_service"), "aoss")
class _FakeGraphResponse:
def __init__(self, status_code, payload=None, text=""):
self.status_code = status_code
@@ -880,5 +878,61 @@ scopes = https://www.googleapis.com/auth/gmail.modify
mock_gmail_connection.call_args.kwargs.get("service_account_user"),
"delegated@example.com",
)
class TestImapFallbacks(unittest.TestCase):
def testDeleteSuccessDoesNotUseFallback(self):
connection = IMAPConnection.__new__(IMAPConnection)
connection._client = MagicMock()
connection.delete_message(42)
connection._client.delete_messages.assert_called_once_with([42])
connection._client.add_flags.assert_not_called()
connection._client.expunge.assert_not_called()
def testDeleteFallbackUsesFlagsAndExpunge(self):
connection = IMAPConnection.__new__(IMAPConnection)
connection._client = MagicMock()
connection._client.delete_messages.side_effect = IMAPClientError("uid expunge")
connection.delete_message(42)
connection._client.add_flags.assert_called_once_with(
[42], [r"\Deleted"], silent=True
)
connection._client.expunge.assert_called_once_with()
def testDeleteFallbackErrorPropagates(self):
connection = IMAPConnection.__new__(IMAPConnection)
connection._client = MagicMock()
connection._client.delete_messages.side_effect = IMAPClientError("uid expunge")
connection._client.add_flags.side_effect = IMAPClientError("flag failed")
with self.assertRaises(IMAPClientError):
connection.delete_message(42)
def testMoveSuccessDoesNotUseFallback(self):
connection = IMAPConnection.__new__(IMAPConnection)
connection._client = MagicMock()
with patch.object(connection, "delete_message") as delete_mock:
connection.move_message(99, "Archive")
connection._client.move_messages.assert_called_once_with([99], "Archive")
connection._client.copy.assert_not_called()
delete_mock.assert_not_called()
def testMoveFallbackCopiesThenDeletes(self):
connection = IMAPConnection.__new__(IMAPConnection)
connection._client = MagicMock()
connection._client.move_messages.side_effect = IMAPClientError("move failed")
with patch.object(connection, "delete_message") as delete_mock:
connection.move_message(99, "Archive")
connection._client.copy.assert_called_once_with([99], "Archive")
delete_mock.assert_called_once_with(99)
def testMoveFallbackCopyErrorPropagates(self):
connection = IMAPConnection.__new__(IMAPConnection)
connection._client = MagicMock()
connection._client.move_messages.side_effect = IMAPClientError("move failed")
connection._client.copy.side_effect = IMAPClientError("copy failed")
with patch.object(connection, "delete_message") as delete_mock:
with self.assertRaises(IMAPClientError):
connection.move_message(99, "Archive")
delete_mock.assert_not_called()
if __name__ == "__main__":
unittest.main(verbosity=2)