from __future__ import annotations import unittest from unittest.mock import patch from govoplan_mail.backend.config import ImapConfig from govoplan_mail.backend.sending.imap import ( ImapAppendError, ImapConfigurationError, _decode_mailbox_name, _encode_mailbox_name, _extract_mailbox_name, _list_imap_folders_on_client, _quote_mailbox_name, _select_readonly, append_message_to_sent, ) class MailboxClient: utf8_enabled = False capabilities = ("IMAP4REV1", "UTF8=ACCEPT") untagged_responses = {"EXISTS": [b"2"], "UIDVALIDITY": [b"1"]} def __init__(self, listing=None): self.listing = listing or [] self.list_calls = 0 self.status_calls = [] self.select_calls = [] self.append_calls = [] self.logged_out = False def list(self): self.list_calls += 1 return "OK", self.listing def status(self, mailbox, items): self.status_calls.append((mailbox, items)) return "OK", [b'"mailbox" (MESSAGES 2 UNSEEN 1)'] def select(self, mailbox, readonly=False): self.select_calls.append((mailbox, readonly)) return "OK", [b"2"] def response(self, code): return "OK", [b"1"] if code == "UIDVALIDITY" else [] def append(self, mailbox, flags, date_time, message): self.append_calls.append((mailbox, flags, message)) return "OK", [b"APPEND complete"] def logout(self): self.logged_out = True return "BYE", [] class ImapMailboxEncodingTests(unittest.TestCase): def test_rfc_and_real_provider_vectors_round_trip(self): # RFC 3501 section 5.1.3 plus the reported German folder and UTF-16 # surrogate pairs. A plain '+' is not a shift in modified UTF-7. for display, wire in ( ("INBOX", "INBOX"), ("Entwürfe", "Entw&APw-rfe"), ("R&D", "R&-D"), ("+Plus & Sons", "+Plus &- Sons"), ("~peter/mail/台北/日本語", "~peter/mail/&U,BTFw-/&ZeVnLIqe-"), ("📨", "&2D3c6A-"), ('A \\ "B"', 'A \\ "B"'), ): with self.subTest(display=display): self.assertEqual(_encode_mailbox_name(display), wire) self.assertEqual(_decode_mailbox_name(wire), display) def test_list_decodes_names_before_standard_folder_detection_and_status(self): client = MailboxClient([ b'(\\HasNoChildren) "/" "Entw&APw-rfe"', b'(\\HasNoChildren) "/" "Gel&APY-scht"', b'(\\HasNoChildren) "/" "R&-D"', ]) result = _list_imap_folders_on_client( client, host="imap.example.org", port=993, security="tls", include_status=True, ) self.assertEqual([folder.name for folder in result.folders], ["Entwürfe", "Gelöscht", "R&D"]) self.assertEqual(result.detected_folder_mappings, {"drafts": "Entwürfe", "trash": "Gelöscht"}) self.assertTrue(all(folder.message_count == 2 and folder.unseen_count == 1 for folder in result.folders)) self.assertEqual([call[0] for call in client.status_calls], [ '"Entw&APw-rfe"', '"Gel&APY-scht"', '"R&-D"', ]) self.assertEqual(client.list_calls, 1) def test_literal_names_are_decoded_without_stripping_or_unquoting_content(self): for wire, expected in ( (b"Entw&APw-rfe", "Entwürfe"), (b'"R&-D" ', '"R&D" '), ): with self.subTest(wire=wire): line = b'(\\Drafts) "/" {' + str(len(wire)).encode("ascii") + b"}" self.assertEqual(_extract_mailbox_name((line, wire)), (expected, {"\\drafts"})) self.assertIsNone(_extract_mailbox_name(b"")) self.assertIsNone(_extract_mailbox_name(None)) def test_list_rejects_wrong_literal_lengths_and_invalid_provider_encoding(self): with self.assertRaisesRegex(ImapAppendError, "literal"): _extract_mailbox_name((b'() "/" {99}', b"INBOX")) for wire in (b"&APw", b"&!bad-", b"&AGE-", b"&AA-", b"&APx-", b"&2AA-", b"Entw\xffrfe"): with self.subTest(wire=wire), self.assertRaisesRegex(ImapAppendError, "encoding"): _extract_mailbox_name(b'() "/" "' + wire + b'"') def test_select_encodes_unicode_and_quotes_protocol_metacharacters(self): client = MailboxClient() folder = 'Entwürfe / R&D / "Q" \\' self.assertEqual(_select_readonly(client, folder), (2, "1")) self.assertEqual(client.select_calls, [('"Entw&APw-rfe / R&-D / \\"Q\\" \\\\"', True)]) self.assertEqual(client.list_calls, 0) def test_quoted_name_escaping_round_trips_independently_of_charset_encoding(self): for folder in ('Entwürfe "R&D"', ' \\"quoted"\\ ', 'R&D', '📨/日本語', 'back\\slash'): with self.subTest(folder=folder): quoted = _quote_mailbox_name(folder) self.assertEqual(_extract_mailbox_name('() "/" ' + quoted), (folder, set())) def test_saved_wire_names_resolve_without_double_encoding(self): client = MailboxClient([b'() "/" "Entw&APw-rfe"']) _select_readonly(client, "Entw&APw-rfe") _select_readonly(client, "Entwürfe") self.assertEqual(client.select_calls, [('"Entw&APw-rfe"', True)] * 2) self.assertEqual(client.list_calls, 1) def test_literal_name_wins_when_legacy_alias_is_ambiguous(self): client = MailboxClient([ b'() "/" "Entw&APw-rfe"', b'() "/" "Entw&-APw-rfe"', ]) _select_readonly(client, "Entw&APw-rfe") self.assertEqual(client.select_calls, [('"Entw&-APw-rfe"', True)]) def test_saved_ampersand_and_literal_ampersand_remain_distinct(self): client = MailboxClient([b'() "/" "R&-D"']) _select_readonly(client, "R&-D") _select_readonly(client, "R&D") self.assertEqual(client.select_calls, [('"R&-D"', True)] * 2) def test_provider_wire_form_is_preserved_on_the_listed_connection(self): client = MailboxClient([b'() "/" "&U,BTFw-&ZeVnLIqe-"']) result = _list_imap_folders_on_client( client, host="imap.example.org", port=993, security="tls", include_status=True, ) self.assertEqual(result.folders[0].name, "台北日本語") self.assertEqual(client.status_calls[0][0], '"&U,BTFw-&ZeVnLIqe-"') def test_utf8_mode_preserves_literal_ampersands_and_does_not_decode_again(self): client = MailboxClient(['() "/" "Entwürfe &APw-"'.encode("utf-8")]) client.utf8_enabled = True result = _list_imap_folders_on_client( client, host="imap.example.org", port=993, security="tls", include_status=True, ) self.assertEqual(result.folders[0].name, "Entwürfe &APw-") _select_readonly(client, result.folders[0].name) self.assertEqual(client.select_calls, [('"Entwürfe &APw-"', True)]) self.assertEqual(client.status_calls[0][0], '"Entwürfe &APw-"') def test_utf8_capability_alone_does_not_change_the_encoding(self): client = MailboxClient() self.assertEqual(_quote_mailbox_name("Entwürfe", client=client), '"Entw&APw-rfe"') with self.assertRaisesRegex(ImapAppendError, "encoding"): _extract_mailbox_name('() "/" "Entwürfe"'.encode("utf-8")) def test_changing_utf8_mode_invalidates_cached_wire_names(self): client = MailboxClient([b'() "/" "Entw&APw-rfe"']) _list_imap_folders_on_client( client, host="imap.example.org", port=993, security="tls", include_status=False, ) client.utf8_enabled = True self.assertEqual(_quote_mailbox_name("Entwürfe", client=client), '"Entwürfe"') def test_select_rejects_control_characters_before_sending_any_command(self): for folder in ("INBOX\r\nLOGOUT", "bad\x00folder", "bad\x7ffolder", "bad\u2028folder"): client = MailboxClient() with self.subTest(folder=folder), self.assertRaisesRegex(ImapConfigurationError, "control"): _select_readonly(client, folder) self.assertEqual(client.select_calls, []) def test_append_auto_detects_unicode_sent_folder_and_uses_original_wire_name(self): client = MailboxClient([b'(\\Sent) "/" "Gesendet &APw-"']) config = ImapConfig(host="imap.example.org", sent_folder="auto") with patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client): result = append_message_to_sent(b"Subject: test\r\n\r\nBody", imap_config=config) self.assertEqual(result.folder, "Gesendet ü") self.assertEqual(client.append_calls[0][0], '"Gesendet &APw-"') self.assertTrue(client.logged_out) def test_append_explicit_unicode_and_saved_wire_names_target_the_same_mailbox(self): for folder in ("Entwürfe", "Entw&APw-rfe"): with self.subTest(folder=folder): client = MailboxClient([b'() "/" "Entw&APw-rfe"']) config = ImapConfig(host="imap.example.org", sent_folder=folder) with patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client): append_message_to_sent(b"Subject: test\r\n\r\nBody", imap_config=config) self.assertEqual(client.append_calls[0][0], '"Entw&APw-rfe"') def test_failed_legacy_lookup_never_selects_or_appends_to_a_guessed_mailbox(self): client = MailboxClient() with patch.object(client, "list", return_value=("NO", [b"Denied"])), self.assertRaisesRegex( ImapAppendError, "resolving a saved folder name", ): _select_readonly(client, "Entw&APw-rfe") self.assertEqual(client.select_calls, []) def test_append_saved_wire_name_failure_is_not_an_unknown_append_outcome(self): client = MailboxClient() config = ImapConfig(host="imap.example.org", sent_folder="Entw&APw-rfe") with ( patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client), patch.object(client, "list", side_effect=OSError("connection lost")), self.assertRaises(ImapAppendError) as caught, ): append_message_to_sent(b"Subject: test\r\n\r\nBody", imap_config=config) self.assertTrue(caught.exception.temporary) self.assertFalse(caught.exception.outcome_unknown) self.assertEqual(client.append_calls, []) self.assertTrue(client.logged_out) if __name__ == "__main__": unittest.main()