"""IMAP response parsing (SPEC §5.1 provider), using real home.pl wire samples.""" from __future__ import annotations from datetime import datetime from app.providers.imap_smtp.parsing import ( parse_fetch_lines, parse_internaldate, parse_list_line, parse_search_uids, parse_select_untagged, parse_vanished, ) def test_parse_list_special_use() -> None: assert parse_list_line(rb'(\HasNoChildren \UnMarked \Sent) "/" SENT') is not None cases = { rb'(\HasNoChildren \UnMarked \Sent) "/" SENT': ("SENT", "sent"), rb'(\HasNoChildren \UnMarked \Drafts) "/" DRAFTS': ("DRAFTS", "drafts"), rb'(\HasNoChildren \UnMarked \Trash) "/" TRASH': ("TRASH", "trash"), rb'(\HasNoChildren \UnMarked \Junk) "/" SPAM': ("SPAM", "junk"), rb'(\HasNoChildren) "/" INBOX': ("INBOX", "inbox"), } for line, (path, special) in cases.items(): parsed = parse_list_line(line) assert parsed is not None assert parsed.path == path assert parsed.special_use == special def test_parse_list_plain_folder_has_no_special_use() -> None: parsed = parse_list_line(rb'(\HasNoChildren) "/" "Work Stuff"') assert parsed is not None assert parsed.path == "Work Stuff" assert parsed.special_use is None def test_parse_list_skips_noselect_containers() -> None: # \Noselect / \NonExistent hierarchy nodes can't be SELECTed (Gmail's [Gmail] # root, a parent label like "Serial numbers " that only groups children). assert parse_list_line(rb'(\HasChildren \Noselect) "/" "[Gmail]"') is None assert parse_list_line(rb'(\HasChildren \Noselect) "/" "Serial numbers "') is None assert parse_list_line(rb'(\NonExistent \HasChildren) "/" "Parent"') is None # A selectable child of that container is still parsed. child = parse_list_line(rb'(\HasNoChildren) "/" "Serial numbers / licencje"') assert child is not None and child.path == "Serial numbers / licencje" def test_parse_list_utf7_name() -> None: # Modified UTF-7: "&AOk-" decodes to "é" -> folder "Réponses" parsed = parse_list_line(rb'(\HasNoChildren) "/" "R&AOk-ponses"') assert parsed is not None assert parsed.path == "Réponses" def test_parse_gm_labels() -> None: from app.providers.imap_smtp.parsing import parse_gm_labels # System labels (backslash-prefixed) are dropped; custom labels kept + decoded. header = rb'1 FETCH (UID 5 X-GM-LABELS (\Inbox \Important "Konkurs" "Projekty/2026") FLAGS ())' assert parse_gm_labels(header) == ("Konkurs", "Projekty/2026") # Unquoted single-word custom label. assert parse_gm_labels(rb"1 FETCH (UID 5 X-GM-LABELS (\Sent PUP))") == ("PUP",) # Modified UTF-7 decoding of a non-ASCII label ("Zg&AUI-oszenia" -> "Zgłoszenia"). assert parse_gm_labels(rb'1 FETCH (X-GM-LABELS ("Zg&AUI-oszenia"))') == ("Zgłoszenia",) # No X-GM-LABELS at all. assert parse_gm_labels(rb"1 FETCH (UID 5 FLAGS (\Seen))") == () # Only system labels -> empty. assert parse_gm_labels(rb"1 FETCH (X-GM-LABELS (\Inbox \Starred))") == () def test_parse_fetch_lines_carries_gm_labels() -> None: lines = [ rb'1 FETCH (UID 9 X-GM-LABELS (\Inbox "Work") FLAGS (\Seen) ' rb'INTERNALDATE "27-May-2026 19:08:09 +0200" RFC822.SIZE 3 BODY[] {3}', b"abc", b")", ] out = parse_fetch_lines(lines) assert len(out) == 1 assert out[0].gmail_labels == ("Work",) assert out[0].raw == b"abc" def test_parse_internaldate() -> None: dt = parse_internaldate("27-May-2026 19:08:09 +0200") assert dt == datetime.fromisoformat("2026-05-27T19:08:09+02:00") # single-digit day (space padded) assert parse_internaldate(" 1-Jan-2026 05:06:07 +0000") is not None assert parse_internaldate("garbage") is None def test_parse_search_uids() -> None: assert parse_search_uids([b"1 2 3 4 5 8", b"Search completed."]) == [1, 2, 3, 4, 5, 8] assert parse_search_uids([b""]) == [] assert parse_search_uids([]) == [] def test_parse_vanished() -> None: # aioimaplib strips the leading "* " before buffering the line. assert parse_vanished([b"VANISHED (EARLIER) 300:305,405"]) == [ 300, 301, 302, 303, 304, 305, 405, ] # noqa: E501 # Live-expunge form (no EARLIER) and a descending range are both handled. assert parse_vanished([b"VANISHED 41", b"VANISHED (EARLIER) 5:3"]) == [41, 3, 4, 5] # Non-VANISHED lines are ignored; non-bytes entries are skipped. assert parse_vanished([b"12 FETCH (UID 12 FLAGS (\\Seen))", b"OK done"]) == [] assert parse_vanished([]) == [] def test_parse_select_untagged() -> None: lines = [ b"6 EXISTS", b"0 RECENT", b"OK [UIDVALIDITY 1772887273] UIDs valid", b"OK [UIDNEXT 9] Predicted next UID", b"OK [HIGHESTMODSEQ 31] Highest", ] info = parse_select_untagged(lines) assert info["uidvalidity"] == 1772887273 assert info["uidnext"] == 9 assert info["highestmodseq"] == 31 assert info["exists"] == 6 def test_parse_fetch_body() -> None: raw = b"From: a@b.com\r\nSubject: Hi\r\n\r\nBody" header = ( b"6 FETCH (UID 8 FLAGS (\\Answered \\Seen $label1) " b'INTERNALDATE "27-May-2026 19:08:09 +0200" RFC822.SIZE 34 BODY[] {' + str(len(raw)).encode() + b"}" ) lines: list[bytes | bytearray] = [header, bytearray(raw), b")", b"Fetch completed."] parsed = parse_fetch_lines(lines) assert len(parsed) == 1 pf = parsed[0] assert pf.uid == 8 assert pf.raw == raw assert "\\Seen" in pf.flags assert "\\Answered" in pf.flags assert pf.internal_date is not None def test_parse_fetch_flags_only() -> None: lines: list[bytes | bytearray] = [ b"1 FETCH (UID 5 FLAGS (\\Seen))", b"2 FETCH (UID 6 FLAGS (\\Flagged \\Seen))", b"Fetch completed.", ] parsed = parse_fetch_lines(lines) assert [p.uid for p in parsed] == [5, 6] assert parsed[0].raw is None assert parsed[1].flags == frozenset({"\\Flagged", "\\Seen"}) def test_parse_fetch_multiple_bodies() -> None: raw1 = b"msg-one" raw2 = b"message-two-longer" lines: list[bytes | bytearray] = [ b'1 FETCH (UID 1 FLAGS () INTERNALDATE "1-Jan-2026 00:00:00 +0000" BODY[] {%d}' % len(raw1), bytearray(raw1), b")", b'2 FETCH (UID 2 FLAGS (\\Seen) INTERNALDATE "2-Jan-2026 00:00:00 +0000" BODY[] {%d}' % len(raw2), bytearray(raw2), b")", b"Fetch completed.", ] parsed = parse_fetch_lines(lines) assert [p.uid for p in parsed] == [1, 2] assert parsed[0].raw == raw1 assert parsed[1].raw == raw2