253 lines
8.9 KiB
Python
253 lines
8.9 KiB
Python
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,
|
|
list_imap_messages,
|
|
)
|
|
|
|
|
|
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_mock_message_cursor_preserves_order_and_resets_when_stale(self):
|
|
records = [
|
|
{
|
|
"id": "3",
|
|
"kind": "smtp",
|
|
"raw_eml": "Subject: Three\r\n\r\nBody",
|
|
"size_bytes": 26,
|
|
},
|
|
{
|
|
"id": "2",
|
|
"kind": "smtp",
|
|
"raw_eml": "Subject: Two\r\n\r\nBody",
|
|
"size_bytes": 24,
|
|
},
|
|
]
|
|
config = ImapConfig(host="mock.imap.local")
|
|
|
|
with patch(
|
|
"govoplan_mail.backend.sending.imap.list_records",
|
|
return_value=records,
|
|
):
|
|
page = list_imap_messages(
|
|
imap_config=config,
|
|
after_uid="3",
|
|
expected_uidvalidity="mock-v1",
|
|
limit=1,
|
|
)
|
|
reset_page = list_imap_messages(
|
|
imap_config=config,
|
|
after_uid="3",
|
|
expected_uidvalidity="stale",
|
|
limit=1,
|
|
)
|
|
|
|
self.assertEqual([message.uid for message in page.messages], ["2"])
|
|
self.assertEqual(page.offset, 1)
|
|
self.assertFalse(page.cursor_reset)
|
|
self.assertEqual([message.uid for message in reset_page.messages], ["3"])
|
|
self.assertEqual(reset_page.offset, 0)
|
|
self.assertTrue(reset_page.cursor_reset)
|
|
|
|
def test_real_message_listing_delegates_to_shared_client_path(self):
|
|
class Client:
|
|
logged_out = False
|
|
|
|
def logout(self):
|
|
self.logged_out = True
|
|
|
|
client = Client()
|
|
expected = object()
|
|
config = ImapConfig(host="imap.example.org")
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_mail.backend.sending.imap._open_imap",
|
|
return_value=client,
|
|
),
|
|
patch(
|
|
"govoplan_mail.backend.sending.imap._list_imap_messages_on_client",
|
|
return_value=expected,
|
|
) as list_on_client,
|
|
):
|
|
result = list_imap_messages(
|
|
imap_config=config,
|
|
folder=" Archive ",
|
|
limit=25,
|
|
offset=5,
|
|
after_uid="42",
|
|
expected_uidvalidity="7",
|
|
)
|
|
|
|
self.assertIs(result, expected)
|
|
self.assertTrue(client.logged_out)
|
|
list_on_client.assert_called_once_with(
|
|
client,
|
|
host="imap.example.org",
|
|
port=993,
|
|
security="tls",
|
|
folder="Archive",
|
|
limit=25,
|
|
offset=5,
|
|
after_uid="42",
|
|
expected_uidvalidity="7",
|
|
)
|
|
|
|
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()
|