❤️ TEST-#38: Add coverage for Sync and Restore single-mailbox paths

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, "<abc@example.com>")
+        # Server has no message_ids
+        self.fake.uid.side_effect = lambda *_a, **_kw: ("OK", [b""])
+        result = Restore.restore_mailbox(
+            self.app._session,
+            "INBOX",
+            ["<abc@example.com>"],
+            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, "<abc@example.com>")
+
+        # 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",
+            ["<abc@example.com>"],
+            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",
+            ["<missing@x>"],
+            storage=self.storage,
+        )
+        self.assertEqual(result["uploaded"], 0)
+        self.fake.append.assert_not_called()
+
+    def test_skip_duplicates_disabled(self):
+        _save(self.storage, "<abc@example.com>")
+        # Even when server already has it, skip_duplicates=False uploads
+        result = Restore.restore_mailbox(
+            self.app._session,
+            "INBOX",
+            ["<abc@example.com>"],
+            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())

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)