homelab-codex-ws/packages/kb-mail/tests/test_imap.py

"""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()