from __future__ import annotations import smtplib import unittest from unittest.mock import patch from govoplan_core.security.outbound_http import OutboundHttpBlocked from govoplan_mail.backend.config import SmtpConfig from govoplan_mail.backend.sending.smtp import ( SmtpConfigurationError, SmtpBatchPolicy, SmtpBatchSession, SmtpSendError, _open_smtp, _prepare_smtp_send, _smtp_send_result, ) class SmtpSendHelperTests(unittest.TestCase): def test_real_smtp_connections_honor_deployment_egress_policy(self): config = SmtpConfig(host="smtp.internal", port=587, security="starttls") with patch( "govoplan_mail.backend.sending.smtp.validate_outbound_host", side_effect=OutboundHttpBlocked("private network blocked"), ), self.assertRaisesRegex(SmtpConfigurationError, "private network blocked"): _open_smtp(config) def test_smtp_revalidates_and_pins_at_connection_time(self): config = SmtpConfig(host="smtp.example.test", port=587, security="starttls") public = [(2, 1, 6, "", ("93.184.216.34", 587))] private = [(2, 1, 6, "", ("127.0.0.1", 587))] 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_smtp(config) socket_factory.assert_not_called() def test_prepare_smtp_send_validates_envelope(self): config = SmtpConfig(host="smtp.example.org", port=587, security="starttls") with self.assertRaisesRegex(SmtpConfigurationError, "envelope sender"): _prepare_smtp_send(smtp_config=config, envelope_from="", envelope_recipients=["user@example.org"]) with self.assertRaisesRegex(SmtpConfigurationError, "recipient"): _prepare_smtp_send(smtp_config=config, envelope_from="sender@example.org", envelope_recipients=[""]) def test_prepare_smtp_send_filters_blank_recipients(self): config = SmtpConfig(host="smtp.example.org", port=587, security="starttls") self.assertEqual( _prepare_smtp_send( smtp_config=config, envelope_from="sender@example.org", envelope_recipients=["", "user@example.org"], ), ("smtp.example.org", 587, ["user@example.org"]), ) def test_smtp_send_result_decodes_refused_recipients(self): config = SmtpConfig(host="smtp.example.org", port=587, security="starttls") result = _smtp_send_result( smtp_config=config, host="smtp.example.org", port=587, envelope_from="sender@example.org", envelope_recipients=["ok@example.org", "blocked@example.org"], refused={"blocked@example.org": (550, b"blocked")}, ) self.assertEqual(result.accepted_count, 1) self.assertEqual(result.refused_recipients["blocked@example.org"], (550, "blocked")) def test_batch_preflight_reuses_one_authenticated_connection(self): config = SmtpConfig(host="smtp.example.org", port=587, security="starttls") smtp = _FakeSmtp() with patch("govoplan_mail.backend.sending.smtp._open_smtp", return_value=smtp) as opener: with SmtpBatchSession(config) as batch: first = batch.send(b"first", envelope_from="sender@example.org", envelope_recipients=["one@example.org"]) second = batch.send(b"second", envelope_from="sender@example.org", envelope_recipients=["two@example.org"]) opener.assert_called_once_with(config) self.assertFalse(first.session_reused) self.assertTrue(second.session_reused) self.assertEqual(1, second.connection_sequence) self.assertEqual([b"first", b"second"], smtp.messages) self.assertTrue(smtp.quit_called) def test_batch_reconnects_before_next_message_when_health_check_fails(self): config = SmtpConfig(host="smtp.example.org", port=587, security="starttls") first_smtp = _FakeSmtp(noop_error_on_call=1) second_smtp = _FakeSmtp() policy = SmtpBatchPolicy(reconnect_attempts=1) with patch( "govoplan_mail.backend.sending.smtp._open_smtp", side_effect=[first_smtp, second_smtp], ) as opener: with SmtpBatchSession(config, policy=policy) as batch: batch.send(b"first", envelope_from="sender@example.org", envelope_recipients=["one@example.org"]) result = batch.send(b"second", envelope_from="sender@example.org", envelope_recipients=["two@example.org"]) self.assertEqual(2, opener.call_count) self.assertEqual(2, result.connection_sequence) self.assertEqual(1, result.reconnect_count) self.assertEqual([b"first"], first_smtp.messages) self.assertEqual([b"second"], second_smtp.messages) def test_preflight_retries_a_transient_connection_failure(self): config = SmtpConfig(host="smtp.example.org", port=587, security="starttls") smtp = _FakeSmtp() with patch( "govoplan_mail.backend.sending.smtp._open_smtp", side_effect=[OSError("temporary DNS failure"), smtp], ) as opener: with SmtpBatchSession(config, policy=SmtpBatchPolicy(reconnect_attempts=1)) as batch: result = batch.send(b"message", envelope_from="sender@example.org", envelope_recipients=["one@example.org"]) self.assertEqual(2, opener.call_count) self.assertEqual(1, result.reconnect_count) def test_connection_loss_after_send_starts_is_unknown_and_never_replayed(self): config = SmtpConfig(host="smtp.example.org", port=587, security="starttls") smtp = _FakeSmtp(send_error=smtplib.SMTPServerDisconnected("lost")) with patch("govoplan_mail.backend.sending.smtp._open_smtp", return_value=smtp), self.assertRaises(SmtpSendError) as raised: with SmtpBatchSession(config) as batch: batch.send(b"one", envelope_from="sender@example.org", envelope_recipients=["one@example.org"]) self.assertTrue(raised.exception.outcome_unknown) self.assertTrue(raised.exception.systemic) self.assertEqual("smtp_connection_lost_after_transmission", raised.exception.reason_code) self.assertEqual(1, smtp.send_calls) def test_authentication_preflight_is_systemic_and_blocks_batch(self): config = SmtpConfig(host="smtp.example.org", port=587, security="starttls") error = smtplib.SMTPAuthenticationError(535, b"bad credentials") with patch("govoplan_mail.backend.sending.smtp._open_smtp", side_effect=error), self.assertRaises(SmtpSendError) as raised: SmtpBatchSession(config).preflight() self.assertTrue(raised.exception.systemic) self.assertFalse(raised.exception.temporary) self.assertEqual("preflight", raised.exception.phase) self.assertEqual("smtp_authentication_failed", raised.exception.reason_code) class _FakeSmtp: def __init__(self, *, noop_error_on_call: int | None = None, send_error: BaseException | None = None): self.noop_error_on_call = noop_error_on_call self.send_error = send_error self.noop_calls = 0 self.send_calls = 0 self.messages: list[bytes] = [] self.quit_called = False def noop(self): self.noop_calls += 1 if self.noop_error_on_call == self.noop_calls: raise smtplib.SMTPServerDisconnected("stale") return 250, b"ok" def sendmail(self, _sender, _recipients, message): self.send_calls += 1 if self.send_error is not None: raise self.send_error self.messages.append(message) return {} def send_message(self, message, **_kwargs): return self.sendmail(None, None, message.as_bytes()) def quit(self): self.quit_called = True return 221, b"bye" def close(self): return None if __name__ == "__main__": unittest.main()