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)