govoplan-mail/tests/test_imap_parser.py

from __future__ import annotations

import unittest
from unittest.mock import patch

from govoplan_core.security.outbound_http import OutboundHttpBlocked
from govoplan_mail.backend.config import ImapConfig
from govoplan_mail.backend.sending.imap import (
    ImapAppendError,
    ImapConfigurationError,
    _detect_sent_folder,
    _extract_mailbox_name,
    _fetch_message_by_uid,
    _normalize_mailbox_page,
    _open_imap,
    _paged_descending_sequences,
    _parse_fetch_sequence,
    _select_readonly,
    _sequence_set,
    append_message_to_sent,
)


class ImapFolderParserTests(unittest.TestCase):
    def test_real_imap_connections_honor_deployment_egress_policy(self):
        config = ImapConfig(host="imap.internal", port=993, security="tls")
        with patch(
            "govoplan_mail.backend.sending.imap.validate_outbound_host",
            side_effect=OutboundHttpBlocked("private network blocked"),
        ), self.assertRaisesRegex(ImapConfigurationError, "private network blocked"):
            _open_imap(config)

    def test_imap_revalidates_and_pins_at_connection_time(self):
        config = ImapConfig(host="imap.example.test", port=993, security="tls")
        public = [(2, 1, 6, "", ("93.184.216.34", 993))]
        private = [(2, 1, 6, "", ("127.0.0.1", 993))]
        with patch.dict(
            "os.environ",
            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
        ), patch(
            "govoplan_core.security.outbound_http.socket.getaddrinfo",
            side_effect=(public, private),
        ), patch("govoplan_core.security.outbound_http.socket.socket") as socket_factory, self.assertRaisesRegex(
            OutboundHttpBlocked,
            "non-public network",
        ):
            _open_imap(config)
        socket_factory.assert_not_called()

    def test_extracts_quoted_mailbox_after_quoted_delimiter(self):
        self.assertEqual(
            _extract_mailbox_name(b'(\\HasNoChildren \\Sent) "/" "Sent Items"'),
            ("Sent Items", {"\\hasnochildren", "\\sent"}),
        )

    def test_extracts_atom_mailbox(self):
        self.assertEqual(
            _extract_mailbox_name('(\\HasNoChildren) "/" INBOX'),
            ("INBOX", {"\\hasnochildren"}),
        )

    def test_extracts_nil_delimiter_mailbox(self):
        self.assertEqual(
            _extract_mailbox_name('(\\HasNoChildren) NIL "INBOX.Sent"'),
            ("INBOX.Sent", {"\\hasnochildren"}),
        )

    def test_detects_sent_by_flag_before_name(self):
        self.assertEqual(
            _detect_sent_folder([("Archive", {"\\sent"}), ("Sent", set())]),
            "Archive",
        )

    def test_detects_common_sent_name(self):
        self.assertEqual(
            _detect_sent_folder([("INBOX", set()), ("Gesendet", set())]),
            "Gesendet",
        )


class ImapMessagePaginationTests(unittest.TestCase):
    def test_full_message_fetch_uses_partial_range_and_deployment_limit(self):
        class Client:
            command = ""

            def uid(self, command, uid, fetch_spec):
                del command, uid
                self.command = fetch_spec
                return "OK", [(b"1 (UID 1 RFC822.SIZE 11)", b"12345678901")]

        client = Client()
        with patch.dict("os.environ", {"GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES": "10"}), self.assertRaisesRegex(
            ImapAppendError,
            "deployment limit",
        ):
            _fetch_message_by_uid(client, "1")
        self.assertIn("BODY.PEEK[]<0.11>", client.command)

    def test_paginates_sequence_numbers_newest_first(self):
        self.assertEqual(
            _paged_descending_sequences(120, offset=0, limit=5),
            ["120", "119", "118", "117", "116"],
        )
        self.assertEqual(_paged_descending_sequences(120, offset=118, limit=5), ["2", "1"])
        self.assertEqual(_paged_descending_sequences(120, offset=120, limit=5), [])

    def test_compacts_contiguous_sequence_sets(self):
        self.assertEqual(_sequence_set(["5", "4", "3"]), "3:5")
        self.assertEqual(_sequence_set(["9", "7", "5"]), "5,7,9")

    def test_extracts_fetch_sequence_number(self):
        self.assertEqual(
            _parse_fetch_sequence("42 (UID 123 FLAGS (\\Seen) RFC822.SIZE 100)"),
            "42",
        )
        self.assertIsNone(_parse_fetch_sequence("UID 123 FLAGS ()"))

    def test_normalizes_mailbox_pagination_request(self):
        self.assertEqual(_normalize_mailbox_page(folder="", limit=0, offset=-5), ("INBOX", 1, 0))
        self.assertEqual(_normalize_mailbox_page(folder=" Sent ", limit=500, offset=3), ("Sent", 100, 3))


class ImapMailboxCommandTests(unittest.TestCase):
    def test_select_quotes_mailbox_name_with_spaces(self):
        class Client:
            untagged_responses = {"EXISTS": [b"0"], "UIDVALIDITY": [b"1"]}

            def select(self, mailbox, readonly=False):
                self.mailbox = mailbox
                self.readonly = readonly
                return "OK", [b"0"]

            def response(self, code):
                return "OK", [b"1"] if code == "UIDVALIDITY" else []

        client = Client()

        self.assertEqual(_select_readonly(client, "Gesendete Elemente"), (0, "1"))
        self.assertEqual(client.mailbox, '"Gesendete Elemente"')
        self.assertTrue(client.readonly)

    def test_append_quotes_mailbox_name_with_spaces(self):
        class Client:
            def append(self, mailbox, flags, date_time, message):
                self.mailbox = mailbox
                self.flags = flags
                self.date_time = date_time
                self.message = message
                return "OK", [b"APPEND completed"]

            def logout(self):
                return "BYE", [b"logged out"]

        client = Client()
        config = ImapConfig(host="imap.example.org", username="user", password="secret", 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, folder="Gesendete Elemente")

        self.assertEqual(client.mailbox, '"Gesendete Elemente"')
        self.assertEqual(result.folder, "Gesendete Elemente")
        self.assertEqual(result.bytes_appended, len(b"Subject: test\r\n\r\nBody"))


if __name__ == "__main__":
    unittest.main()