diff --git a/tests/clients/imap/test_restore.py b/tests/clients/imap/test_restore.py new file mode 100644 index 0000000..82a8daa --- /dev/null +++ b/tests/clients/imap/test_restore.py @@ -0,0 +1,139 @@ +"""Tests for Restore.restore_mailbox (single-mailbox path).""" + +import tempfile +from pathlib import Path +from unittest import TestCase +from unittest.mock import patch + +from email_profile import Email, StorageSQLite +from email_profile.clients.imap.restore import ( + Restore, + _ensure_mailbox, + _server_message_ids, +) +from email_profile.serializers.raw import RawSerializer +from tests.conftest import SAMPLE_RFC822, make_fake_client + + +def _save(storage: StorageSQLite, message_id: str, mailbox: str = "INBOX"): + storage.save( + RawSerializer( + message_id=message_id, + uid="1", + mailbox=mailbox, + file=SAMPLE_RFC822, + ) + ) + + +class _RestoreTest(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() + self._tmp = tempfile.TemporaryDirectory() + self.storage = StorageSQLite(Path(self._tmp.name) / "mail.db") + self.restore = Restore(self.app._session) + + def tearDown(self): + self.app.close() + self._patcher.stop() + self._tmp.cleanup() + + +class TestRestoreMailbox(_RestoreTest): + def test_uploads_new_message(self): + _save(self.storage, "") + # Server has no message_ids + self.fake.uid.side_effect = lambda *_a, **_kw: ("OK", [b""]) + result = Restore.restore_mailbox( + self.app._session, + "INBOX", + [""], + storage=self.storage, + ) + self.assertEqual(result["uploaded"], 1) + self.assertEqual(result["skipped"], 0) + self.fake.append.assert_called_once() + + def test_skips_existing_message_ids(self): + _save(self.storage, "") + + # Server returns the same message_id + def side(command, *args): + cmd = command.upper() + if cmd == "SEARCH": + return ("OK", [b"1"]) + if cmd == "FETCH": + header = ( + b"1 (UID 1 BODY.PEEK[HEADER.FIELDS (MESSAGE-ID)] {%d}" + % len(SAMPLE_RFC822) + ) + return ("OK", [(header, SAMPLE_RFC822), b")"]) + return ("OK", [b"Done"]) + + self.fake.uid.side_effect = side + + result = Restore.restore_mailbox( + self.app._session, + "INBOX", + [""], + storage=self.storage, + ) + self.assertEqual(result["uploaded"], 0) + self.assertEqual(result["skipped"], 1) + self.fake.append.assert_not_called() + + def test_missing_storage_entry_dropped(self): + # Storage has nothing for this id + self.fake.uid.side_effect = lambda *_a, **_kw: ("OK", [b""]) + result = Restore.restore_mailbox( + self.app._session, + "INBOX", + [""], + storage=self.storage, + ) + self.assertEqual(result["uploaded"], 0) + self.fake.append.assert_not_called() + + def test_skip_duplicates_disabled(self): + _save(self.storage, "") + # Even when server already has it, skip_duplicates=False uploads + result = Restore.restore_mailbox( + self.app._session, + "INBOX", + [""], + storage=self.storage, + skip_duplicates=False, + ) + self.assertEqual(result["uploaded"], 1) + + +class TestEnsureMailbox(_RestoreTest): + def test_existing_mailbox_no_op(self): + before = self.fake.create.call_count + _ensure_mailbox(self.app._session, "INBOX") + self.assertEqual(self.fake.create.call_count, before) + + def test_missing_mailbox_created(self): + self.fake.list.return_value = ( + "OK", + [ + b'(\\HasNoChildren) "/" "INBOX"', + b'(\\HasNoChildren) "/" "Archive"', + ], + ) + _ensure_mailbox(self.app._session, "Archive") + self.fake.create.assert_called_once() + self.assertIn("Archive", self.app._session.mailboxes) + + +class TestServerMessageIds(_RestoreTest): + def test_returns_empty_on_empty_search(self): + self.fake.uid.side_effect = lambda *_a, **_kw: ("OK", [b""]) + ids = _server_message_ids(self.app._session.client, "INBOX") + self.assertEqual(ids, set()) diff --git a/tests/clients/imap/test_sync.py b/tests/clients/imap/test_sync.py new file mode 100644 index 0000000..f087646 --- /dev/null +++ b/tests/clients/imap/test_sync.py @@ -0,0 +1,130 @@ +"""Tests for Sync.sync (single-mailbox path).""" + +import tempfile +from pathlib import Path +from unittest import TestCase +from unittest.mock import patch + +from email_profile import Email, StorageSQLite +from email_profile.clients.imap.sync import Sync +from tests.conftest import SAMPLE_RFC822, make_fake_client + + +def _fetch_uid_search(uids: list[bytes]) -> dict[str, object]: + """A fake uid() that returns the given UIDs to SEARCH and bodies to FETCH.""" + return list(uids) + + +class _SyncTest(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() + + self._tmp = tempfile.TemporaryDirectory() + self.storage = StorageSQLite(Path(self._tmp.name) / "mail.db") + self.sync = Sync(self.app._session) + + def tearDown(self): + self.app.close() + self._patcher.stop() + self._tmp.cleanup() + + +class TestSyncSingleMailbox(_SyncTest): + def test_inserts_new_messages(self): + # Wire up UID SEARCH → [1], UID FETCH → SAMPLE_RFC822 + def side(command, *args): + cmd = command.upper() + if cmd == "SEARCH": + return ("OK", [b"1"]) + if cmd == "FETCH": + header = b"1 (UID 1 FLAGS (\\Seen) RFC822 {%d}" % len( + SAMPLE_RFC822 + ) + return ("OK", [(header, SAMPLE_RFC822), b")"]) + return ("OK", [b"Done"]) + + self.fake.uid.side_effect = side + + result = self.sync.sync(self.app.inbox, self.storage) + self.assertEqual(result.inserted, 1) + self.assertEqual(result.updated, 0) + self.assertEqual(result.skipped, 0) + self.assertFalse(result.has_errors) + + def test_skips_duplicate_message_ids(self): + def side(command, *args): + cmd = command.upper() + if cmd == "SEARCH": + return ("OK", [b"1"]) + if cmd == "FETCH": + header = b"1 (UID 1 FLAGS () RFC822 {%d}" % len(SAMPLE_RFC822) + return ("OK", [(header, SAMPLE_RFC822), b")"]) + return ("OK", [b"Done"]) + + self.fake.uid.side_effect = side + # First sync stores the message + first = self.sync.sync(self.app.inbox, self.storage) + self.assertEqual(first.inserted, 1) + # Second sync sees the same message-id and skips + second = self.sync.sync(self.app.inbox, self.storage) + self.assertEqual(second.inserted, 0) + self.assertEqual(second.skipped, 1) + + def test_empty_mailbox_no_work(self): + def side(command, *args): + if command.upper() == "SEARCH": + return ("OK", [b""]) + return ("OK", [b"Done"]) + + self.fake.uid.side_effect = side + result = self.sync.sync(self.app.inbox, self.storage) + self.assertEqual(result.inserted, 0) + self.assertEqual(result.updated, 0) + self.assertEqual(result.skipped, 0) + + def test_progress_callback_invoked(self): + def side(command, *args): + cmd = command.upper() + if cmd == "SEARCH": + return ("OK", [b"1"]) + if cmd == "FETCH": + header = b"1 (UID 1 FLAGS () RFC822 {%d}" % len(SAMPLE_RFC822) + return ("OK", [(header, SAMPLE_RFC822), b")"]) + return ("OK", [b"Done"]) + + self.fake.uid.side_effect = side + + events: list[tuple[int, int]] = [] + self.sync.sync( + self.app.inbox, + self.storage, + on_progress=lambda done, total: events.append((done, total)), + ) + self.assertTrue(events) + self.assertEqual(events[-1][0], events[-1][1]) + + def test_storage_save_failure_recorded_as_error(self): + def side(command, *args): + cmd = command.upper() + if cmd == "SEARCH": + return ("OK", [b"1"]) + if cmd == "FETCH": + header = b"1 (UID 1 FLAGS () RFC822 {%d}" % len(SAMPLE_RFC822) + return ("OK", [(header, SAMPLE_RFC822), b")"]) + return ("OK", [b"Done"]) + + self.fake.uid.side_effect = side + + with patch.object( + self.storage, "save", side_effect=RuntimeError("disk full") + ): + result = self.sync.sync(self.app.inbox, self.storage) + + self.assertTrue(result.has_errors) + self.assertEqual(result.inserted, 0)