232 lines
11 KiB
Python
232 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import imaplib
|
|
import unittest
|
|
from dataclasses import replace
|
|
from unittest.mock import Mock, patch
|
|
|
|
from govoplan_mail.backend.config import ImapConfig
|
|
from govoplan_mail.backend.sending.imap import (
|
|
ImapAppendError,
|
|
ImapBatchPolicy,
|
|
ImapBatchSession,
|
|
ImapConfigurationError,
|
|
append_message_to_sent,
|
|
)
|
|
|
|
|
|
class BatchClient:
|
|
utf8_enabled = False
|
|
|
|
def __init__(self, *, wire_folder=b"Gesendete &APw-bermittlung"):
|
|
self.login = Mock(return_value=("OK", []))
|
|
self.list = Mock(return_value=("OK", [b'(\\Sent) "/" "' + wire_folder + b'"']))
|
|
self.noop = Mock(return_value=("OK", []))
|
|
self.append = Mock(return_value=("OK", [b"APPEND complete"]))
|
|
self.logout = Mock(return_value=("BYE", []))
|
|
self.shutdown = Mock()
|
|
|
|
|
|
def config(**changes):
|
|
return ImapConfig(
|
|
host="imap.example.test", port=993, security="tls",
|
|
username="service", password="secret", sent_folder="auto",
|
|
).model_copy(update=changes)
|
|
|
|
|
|
class ImapBatchTests(unittest.TestCase):
|
|
def test_one_login_discovery_and_logout_for_many_sequential_appends(self):
|
|
client = BatchClient()
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap.validate_outbound_host"),
|
|
patch("govoplan_mail.backend.sending.imap._OutboundPolicyIMAP4SSL", return_value=client) as connect,
|
|
ImapBatchSession(config()) as batch,
|
|
):
|
|
results = [append_message_to_sent(bytes([i]), imap_config=config(), batch_session=batch) for i in range(50)]
|
|
self.assertEqual(batch.connection_count, 1)
|
|
self.assertEqual(batch.reconnect_count, 0)
|
|
client.logout.assert_not_called()
|
|
connect.assert_called_once()
|
|
client.login.assert_called_once_with("service", "secret")
|
|
client.list.assert_called_once()
|
|
client.noop.assert_not_called()
|
|
client.logout.assert_called_once()
|
|
self.assertEqual(client.append.call_count, 50)
|
|
self.assertEqual([call.args[3] for call in client.append.call_args_list], [bytes([i]) for i in range(50)])
|
|
self.assertEqual({call.args[0] for call in client.append.call_args_list}, {'"Gesendete &APw-bermittlung"'})
|
|
self.assertEqual({item.folder for item in results}, {"Gesendete übermittlung"})
|
|
self.assertEqual([item.session_reused for item in results], [False] + [True] * 49)
|
|
self.assertEqual({item.connection_sequence for item in results}, {1})
|
|
|
|
def test_count_rotation_rediscovers_original_wire_names_on_new_connection(self):
|
|
first = BatchClient(wire_folder=b"&U,BTFw-&ZeVnLIqe-")
|
|
second = BatchClient(wire_folder=b"&U,BTF2XlZyyKng-")
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[first, second]) as connect,
|
|
ImapBatchSession(config(), policy=replace(ImapBatchPolicy(), max_messages_per_connection=2)) as batch,
|
|
):
|
|
results = [batch.append(b"message") for _ in range(3)]
|
|
self.assertEqual(connect.call_count, 2)
|
|
first.list.assert_called_once()
|
|
second.list.assert_called_once()
|
|
first.logout.assert_called_once()
|
|
second.logout.assert_called_once()
|
|
self.assertEqual([item.connection_sequence for item in results], [1, 1, 2])
|
|
self.assertEqual([item.session_reused for item in results], [False, True, False])
|
|
self.assertEqual(results[-1].reconnect_count, 1)
|
|
self.assertEqual(first.append.call_args.args[0], '"&U,BTFw-&ZeVnLIqe-"')
|
|
self.assertEqual(second.append.call_args.args[0], '"&U,BTF2XlZyyKng-"')
|
|
|
|
def test_age_rotation_is_checked_before_next_append(self):
|
|
clock = [0.0]
|
|
first, second = BatchClient(), BatchClient()
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap.time.monotonic", side_effect=lambda: clock[0]),
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[first, second]),
|
|
ImapBatchSession(config(), policy=replace(ImapBatchPolicy(), max_connection_age_seconds=20)) as batch,
|
|
):
|
|
batch.append(b"one")
|
|
clock[0] = 20.0
|
|
self.assertFalse(batch.append(b"two").session_reused)
|
|
first.noop.assert_not_called()
|
|
self.assertEqual(first.append.call_count + second.append.call_count, 2)
|
|
|
|
def test_idle_probe_can_reconnect_before_append_without_replaying_prior_message(self):
|
|
clock = [0.0]
|
|
first, second = BatchClient(), BatchClient()
|
|
first.noop.side_effect = imaplib.IMAP4.abort("gone")
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap.time.monotonic", side_effect=lambda: clock[0]),
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[first, second]),
|
|
ImapBatchSession(config()) as batch,
|
|
):
|
|
batch.append(b"one")
|
|
clock[0] = 31.0
|
|
result = batch.append(b"two")
|
|
first.noop.assert_called_once()
|
|
self.assertEqual(first.append.call_args.args[3], b"one")
|
|
self.assertEqual(second.append.call_args.args[3], b"two")
|
|
self.assertEqual(result.reconnect_count, 1)
|
|
|
|
def test_healthy_idle_probe_reuses_connection(self):
|
|
client = BatchClient()
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client),
|
|
ImapBatchSession(config(), policy=replace(ImapBatchPolicy(), idle_health_check_seconds=0)) as batch,
|
|
):
|
|
batch.append(b"one")
|
|
self.assertTrue(batch.append(b"two").session_reused)
|
|
client.noop.assert_called_once()
|
|
|
|
def test_only_pre_effect_connection_failures_are_retried(self):
|
|
client = BatchClient()
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[OSError("offline"), client]) as connect,
|
|
ImapBatchSession(config()) as batch,
|
|
):
|
|
result = batch.append(b"one")
|
|
self.assertEqual(connect.call_count, 2)
|
|
client.append.assert_called_once()
|
|
self.assertEqual(result.reconnect_count, 1)
|
|
|
|
def test_pre_effect_reconnects_are_bounded_and_not_unknown(self):
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=OSError("offline")) as connect,
|
|
ImapBatchSession(config()) as batch,
|
|
self.assertRaises(ImapAppendError) as caught,
|
|
):
|
|
batch.append(b"one")
|
|
self.assertEqual(connect.call_count, 2)
|
|
self.assertTrue(caught.exception.temporary)
|
|
self.assertFalse(caught.exception.outcome_unknown)
|
|
|
|
def test_authentication_rejection_is_not_retried(self):
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=imaplib.IMAP4.error("bad login")) as connect,
|
|
ImapBatchSession(config()) as batch,
|
|
self.assertRaises(ImapAppendError) as caught,
|
|
):
|
|
batch.append(b"one")
|
|
connect.assert_called_once()
|
|
self.assertFalse(caught.exception.temporary)
|
|
self.assertFalse(caught.exception.outcome_unknown)
|
|
|
|
def test_ambiguous_append_is_never_replayed_and_connection_is_discarded(self):
|
|
first, second = BatchClient(), BatchClient()
|
|
first.append.side_effect = imaplib.IMAP4.abort("accepted but reply lost")
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=[first, second]) as connect,
|
|
ImapBatchSession(config()) as batch,
|
|
):
|
|
with self.assertRaises(ImapAppendError) as caught:
|
|
batch.append(b"uncertain")
|
|
self.assertTrue(caught.exception.outcome_unknown)
|
|
self.assertFalse(caught.exception.temporary)
|
|
self.assertEqual(connect.call_count, 1)
|
|
first.logout.assert_called_once()
|
|
batch.append(b"different independently claimed message")
|
|
first.append.assert_called_once()
|
|
second.append.assert_called_once()
|
|
self.assertNotEqual(first.append.call_args.args[3], second.append.call_args.args[3])
|
|
|
|
def test_definitive_append_rejection_is_not_retried_or_unknown(self):
|
|
client = BatchClient()
|
|
client.append.return_value = ("NO", [b"quota"])
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client) as connect,
|
|
ImapBatchSession(config()) as batch,
|
|
self.assertRaises(ImapAppendError) as caught,
|
|
):
|
|
batch.append(b"one")
|
|
self.assertFalse(caught.exception.outcome_unknown)
|
|
connect.assert_called_once()
|
|
client.append.assert_called_once()
|
|
|
|
def test_logout_failure_does_not_reverse_accepted_message(self):
|
|
client = BatchClient()
|
|
client.logout.side_effect = OSError("gone")
|
|
with patch("govoplan_mail.backend.sending.imap._open_imap", return_value=client):
|
|
result = append_message_to_sent(b"one", imap_config=config())
|
|
self.assertEqual(result.bytes_appended, 3)
|
|
client.shutdown.assert_called_once()
|
|
|
|
def test_disabling_reuse_keeps_individual_appends_and_closes_every_connection(self):
|
|
clients = [BatchClient() for _ in range(3)]
|
|
with (
|
|
patch("govoplan_mail.backend.sending.imap._open_imap", side_effect=clients),
|
|
ImapBatchSession(config(), policy=replace(ImapBatchPolicy(), reuse_connections=False)) as batch,
|
|
):
|
|
results = [batch.append(b"one") for _ in clients]
|
|
self.assertEqual([item.connection_sequence for item in results], [1, 2, 3])
|
|
self.assertFalse(any(item.session_reused for item in results))
|
|
for client in clients:
|
|
client.append.assert_called_once()
|
|
client.logout.assert_called_once()
|
|
|
|
def test_config_mismatch_closed_and_overlapping_calls_fail_before_provider(self):
|
|
with patch("govoplan_mail.backend.sending.imap._open_imap") as connect:
|
|
batch = ImapBatchSession(config())
|
|
with self.assertRaises(ImapConfigurationError):
|
|
append_message_to_sent(b"one", imap_config=config(password="different"), batch_session=batch)
|
|
with batch._exclusive_append(), self.assertRaises(ImapConfigurationError):
|
|
batch.append(b"overlap")
|
|
batch.close()
|
|
with self.assertRaises(ImapConfigurationError):
|
|
batch.append(b"closed")
|
|
connect.assert_not_called()
|
|
|
|
def test_environment_values_are_bounded(self):
|
|
with patch.dict("os.environ", {
|
|
"GOVOPLAN_IMAP_BATCH_REUSE": "false",
|
|
"GOVOPLAN_IMAP_BATCH_MAX_MESSAGES": "0",
|
|
"GOVOPLAN_IMAP_BATCH_MAX_AGE_SECONDS": "invalid",
|
|
"GOVOPLAN_IMAP_BATCH_IDLE_HEALTH_CHECK_SECONDS": "999999",
|
|
"GOVOPLAN_IMAP_BATCH_RECONNECT_ATTEMPTS": "999999",
|
|
}):
|
|
policy = ImapBatchPolicy.from_environment()
|
|
self.assertFalse(policy.reuse_connections)
|
|
self.assertEqual(policy.max_messages_per_connection, 1)
|
|
self.assertEqual(policy.max_connection_age_seconds, 300)
|
|
self.assertEqual(policy.idle_health_check_seconds, 3600)
|
|
self.assertEqual(policy.reconnect_attempts, 5)
|