"""Unit tests for the IMAP adapter — no network, no real imaplib connection. `ImapClient` is exercised against `_FakeIMAP`, a stand-in for `imaplib.IMAP4_SSL` that speaks real IMAP response shapes (bytes lines, `(prefix, literal)` FETCH tuples, untagged responses). The point is to test this module's OWN parsing rather than a mock of it — every response shape here is one imaplib actually hands back. """ from __future__ import annotations from datetime import date import pytest from kb_mail.imap import ( DEFAULT_PORT, ImapAccount, ImapClient, ImapError, imap_date, parse_list_line, quote_mailbox, ) class _FakeIMAP: """Minimal IMAP4_SSL stand-in. Records commands so tests can assert on the wire traffic — notably that folders are opened READ-ONLY and bodies fetched with BODY.PEEK[].""" def __init__( self, *, list_data=None, messages=None, uidvalidity=42, uidnext=101, status_line=None, search_ok=True, select_ok=True, untagged=True, ): self.list_data = list_data if list_data is not None else [ b'(\\HasNoChildren) "/" "INBOX"', b'(\\HasNoChildren \\All) "/" "[Gmail]/Wszystkie"', b'(\\HasNoChildren \\Sent) "/" "[Gmail]/Wys&AWI-ane"', ] self.messages = messages or {} self.uidvalidity = uidvalidity self.uidnext = uidnext self.status_line = status_line self.search_ok = search_ok self.select_ok = select_ok self.untagged = untagged self.commands: list[tuple] = [] self.logged_in = False self.logged_out = False def login(self, user, password): self.commands.append(("LOGIN", user)) self.logged_in = True return ("OK", [b"LOGIN completed"]) def logout(self): self.logged_out = True return ("BYE", [b"logging out"]) def list(self, directory='""', pattern="*"): self.commands.append(("LIST",)) return ("OK", self.list_data) def select(self, mailbox, readonly=False): self.commands.append(("SELECT", mailbox, readonly)) if not self.select_ok: return ("NO", [b"no such mailbox"]) return ("OK", [str(len(self.messages)).encode()]) def status(self, mailbox, names): self.commands.append(("STATUS", mailbox, names)) line = self.status_line or ( f'"{mailbox}" (MESSAGES {len(self.messages)} ' f"UIDNEXT {self.uidnext} UIDVALIDITY {self.uidvalidity})" ).encode() return ("OK", [line]) def response(self, key): if not self.untagged: return (key, [None]) if key == "UIDVALIDITY": return (key, [str(self.uidvalidity).encode()]) if key == "UIDNEXT": return (key, [str(self.uidnext).encode()]) return (key, [None]) def uid(self, command, *args): self.commands.append(("UID", command) + args) if command == "SEARCH": if not self.search_ok: return ("NO", [b"search failed"]) criteria = args[1:] return ("OK", [self._search(criteria)]) if command == "FETCH": uid = int(args[0]) raw = self.messages.get(uid) if raw is None: return ("OK", [None]) prefix = f"1 (UID {uid} BODY[] {{{len(raw)}}}".encode() return ("OK", [(prefix, raw), b")"]) raise AssertionError(f"unexpected UID command {command}") def _search(self, criteria): uids = sorted(self.messages) if criteria and criteria[0] == "UID": start = int(criteria[1].split(":")[0]) hits = [u for u in uids if u >= start] # Real servers resolve `n:*` as a range and return the highest message even when # n is beyond it. Reproduced deliberately — the client-side filter exists for this. if not hits and uids: hits = [uids[-1]] return " ".join(str(u) for u in hits).encode() return " ".join(str(u) for u in uids).encode() def _account(**kwargs) -> ImapAccount: base = dict(name="gmail", host="imap.example.com", user="u@example.com", password="secret", special_use=("\\All",)) base.update(kwargs) return ImapAccount(**base) def _client(fake: _FakeIMAP, account=None) -> ImapClient: account = account or _account() client = ImapClient(account, connection_factory=lambda acc, timeout: fake) client.connect() return client class TestImapAccount: def test_defaults_to_implicit_tls_port(self): assert _account().port == DEFAULT_PORT def test_password_is_not_in_repr(self): # The repo has two documented secret leaks to session transcripts (recon Decyzja (c)). assert "secret" not in repr(_account()) def test_rejects_account_with_no_scope(self): with pytest.raises(ValueError, match="no scope"): ImapAccount(name="x", host="h", user="u", password="p") def test_rejects_unknown_initial_mode(self): with pytest.raises(ValueError, match="initial_mode"): _account(initial_mode="yesterday") def test_since_mode_requires_a_date(self): with pytest.raises(ValueError, match="initial_since"): _account(initial_mode="since") def test_since_mode_with_date_is_valid(self): assert _account(initial_mode="since", initial_since=date(2026, 6, 15)).initial_since class TestHelpers: def test_imap_date_uses_english_month_abbreviations(self): assert imap_date(date(2026, 6, 15)) == "15-Jun-2026" def test_imap_date_zero_pads_the_day(self): assert imap_date(date(2026, 12, 5)) == "05-Dec-2026" def test_quote_mailbox_wraps_names_with_spaces(self): assert quote_mailbox("[Gmail]/All Mail") == '"[Gmail]/All Mail"' def test_quote_mailbox_escapes_quotes_and_backslashes(self): assert quote_mailbox('we"ird\\name') == '"we\\"ird\\\\name"' class TestParseListLine: def test_quoted_name_with_attributes(self): attrs, name = parse_list_line(b'(\\HasNoChildren \\All) "/" "[Gmail]/All Mail"') assert name == "[Gmail]/All Mail" assert "\\all" in attrs def test_unquoted_atom_name(self): attrs, name = parse_list_line(b'(\\HasNoChildren) "/" INBOX') assert name == "INBOX" def test_nil_delimiter(self): attrs, name = parse_list_line(b'(\\Noselect) NIL "Archive"') assert name == "Archive" def test_literal_name_arrives_as_a_tuple(self): attrs, name = parse_list_line((b'(\\HasNoChildren \\Archive) "/" {7}', b"Archiwa")) assert name == "Archiwa" assert "\\archive" in attrs def test_unparseable_line_returns_none(self): assert parse_list_line(b"* SOMETHING ELSE") is None class TestResolveFolders: def test_special_use_is_matched_by_attribute_not_name(self): # The Polish-UI name is what makes hardcoding '[Gmail]/All Mail' a silent zero-mail sync. fake = _FakeIMAP() assert _client(fake).resolve_folders() == ["[Gmail]/Wszystkie"] def test_literal_folders_are_used_as_configured(self): fake = _FakeIMAP() account = _account(name="fastmail", special_use=(), folders=("INBOX", "Archive", "Sent")) assert _client(fake, account).resolve_folders() == ["INBOX", "Archive", "Sent"] def test_missing_special_use_raises_instead_of_syncing_nothing(self): fake = _FakeIMAP(list_data=[b'(\\HasNoChildren) "/" "INBOX"']) with pytest.raises(ImapError, match="SPECIAL-USE"): _client(fake).resolve_folders() def test_duplicates_between_attribute_and_literal_are_collapsed(self): fake = _FakeIMAP() account = _account(special_use=("\\All",), folders=("[Gmail]/Wszystkie", "INBOX")) assert _client(fake, account).resolve_folders() == ["[Gmail]/Wszystkie", "INBOX"] def test_no_list_call_when_only_literal_folders_are_configured(self): fake = _FakeIMAP() account = _account(special_use=(), folders=("INBOX",)) _client(fake, account).resolve_folders() assert not any(c[0] == "LIST" for c in fake.commands) class TestExamine: def test_opens_folder_read_only(self): # A plain SELECT + FETCH RFC822 would set \Seen on the operator's mail. fake = _FakeIMAP(messages={1: b"raw"}) _client(fake).examine("INBOX") select_cmds = [c for c in fake.commands if c[0] == "SELECT"] assert select_cmds and select_cmds[0][2] is True def test_returns_uidvalidity_and_uidnext_from_untagged_responses(self): fake = _FakeIMAP(uidvalidity=7, uidnext=55) status = _client(fake).examine("INBOX") assert (status.uidvalidity, status.uidnext) == (7, 55) assert not any(c[0] == "STATUS" for c in fake.commands) def test_falls_back_to_status_when_server_omits_untagged_values(self): fake = _FakeIMAP(uidvalidity=9, uidnext=30, untagged=False) status = _client(fake).examine("INBOX") assert (status.uidvalidity, status.uidnext) == (9, 30) assert any(c[0] == "STATUS" for c in fake.commands) # ... and re-opens the mailbox, since STATUS can deselect it. assert len([c for c in fake.commands if c[0] == "SELECT"]) == 2 def test_non_ok_select_raises(self): fake = _FakeIMAP(select_ok=False) with pytest.raises(ImapError, match="EXAMINE"): _client(fake).examine("Nope") class TestStatus: def test_parses_message_count_for_the_sizing_measurement(self): fake = _FakeIMAP(messages={1: b"a", 2: b"b"}, uidnext=3, uidvalidity=11) status = _client(fake).status("INBOX") assert (status.messages, status.uidnext, status.uidvalidity) == (2, 3, 11) def test_missing_field_raises_rather_than_defaulting(self): fake = _FakeIMAP(status_line=b'"INBOX" (MESSAGES 2)') with pytest.raises(ImapError, match="has no UIDVALIDITY"): _client(fake).status("INBOX") class TestSearch: def test_search_all_returns_sorted_uids(self): fake = _FakeIMAP(messages={3: b"c", 1: b"a", 2: b"b"}) assert _client(fake).search_all() == [1, 2, 3] def test_search_since_sends_an_imap_date(self): fake = _FakeIMAP(messages={1: b"a"}) _client(fake).search_since(date(2026, 6, 15)) search = [c for c in fake.commands if c[:2] == ("UID", "SEARCH")][0] assert search[-2:] == ("SINCE", "15-Jun-2026") def test_search_from_uid_filters_the_range_trailer(self): # `n:*` past the end returns the LAST message on real servers; without the # client-side filter every idle tick would report one "new" mail forever. fake = _FakeIMAP(messages={1: b"a", 2: b"b"}) assert _client(fake).search_from_uid(3) == [] def test_search_from_uid_returns_only_new_uids(self): fake = _FakeIMAP(messages={1: b"a", 2: b"b", 3: b"c"}) assert _client(fake).search_from_uid(2) == [2, 3] def test_non_ok_search_raises(self): fake = _FakeIMAP(messages={1: b"a"}, search_ok=False) with pytest.raises(ImapError, match="UID SEARCH"): _client(fake).search_all() class TestFetch: def test_returns_raw_bytes_and_uses_body_peek(self): fake = _FakeIMAP(messages={5: b"From: a@b\r\n\r\nhello"}) assert _client(fake).fetch_message(5) == b"From: a@b\r\n\r\nhello" fetch = [c for c in fake.commands if c[:2] == ("UID", "FETCH")][0] assert fetch[-1] == "(BODY.PEEK[])" def test_missing_uid_returns_none_not_an_error(self): # Expunged between SEARCH and FETCH — an ordinary race on a live mailbox. fake = _FakeIMAP(messages={5: b"x"}) assert _client(fake).fetch_message(6) is None def test_mismatched_uid_in_response_raises(self): fake = _FakeIMAP(messages={5: b"x"}) client = _client(fake) fake.uid = lambda command, *args: ("OK", [(b"1 (UID 9 BODY[] {1}", b"x"), b")"]) with pytest.raises(ImapError, match="returned UID 9"): client.fetch_message(5) class TestLifecycle: def test_context_manager_logs_in_and_out(self): fake = _FakeIMAP() with ImapClient(_account(), connection_factory=lambda acc, t: fake): assert fake.logged_in assert fake.logged_out def test_logout_runs_even_when_the_body_raises(self): fake = _FakeIMAP() with pytest.raises(RuntimeError): with ImapClient(_account(), connection_factory=lambda acc, t: fake): raise RuntimeError("boom") assert fake.logged_out def test_commands_before_connect_raise(self): client = ImapClient(_account(), connection_factory=lambda acc, t: _FakeIMAP()) with pytest.raises(ImapError, match="not connected"): client.search_all()