from __future__ import annotations import json import unittest from dataclasses import asdict from types import SimpleNamespace from unittest.mock import patch from govoplan_mail.backend.capabilities import ( MailCampaignCapability, append_campaign_message_to_sent, campaign_profile_delivery_summary, send_campaign_email_bytes, ) from govoplan_mail.backend.config import ImapConfig, SmtpConfig from govoplan_mail.backend.mail_profiles import MailProfileError from govoplan_mail.backend.sending.imap import ImapAppendError from govoplan_mail.backend.sending.smtp import SmtpSendError def _profile() -> SimpleNamespace: return SimpleNamespace( smtp_config={ "host": "smtp.internal.example", "port": 587, "security": "starttls", }, smtp_username="service-account", smtp_password_encrypted="encrypted-secret", smtp_transport_revision="smtp-revision", imap_config=None, imap_username=None, imap_password_encrypted=None, imap_transport_revision="imap-revision", ) class CampaignMailCapabilityTests(unittest.TestCase): def test_summary_does_not_materialize_or_decrypt_credentials(self) -> None: profile = _profile() with ( patch( "govoplan_mail.backend.capabilities.ensure_mail_profile_allowed_for_campaign", return_value=profile, ), patch("govoplan_mail.backend.capabilities.effective_mail_profile_policy", return_value=object()), patch("govoplan_mail.backend.capabilities._assert_campaign_inherits_profile_credentials"), patch( "govoplan_mail.backend.capabilities.smtp_config_from_profile", side_effect=AssertionError("summary must not decrypt SMTP credentials"), ), patch( "govoplan_mail.backend.capabilities.imap_config_from_profile", side_effect=AssertionError("summary must not decrypt IMAP credentials"), ), ): summary = campaign_profile_delivery_summary( object(), # type: ignore[arg-type] tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", ) self.assertTrue(summary["smtp_available"]) self.assertFalse(summary["imap_available"]) self.assertNotIn("service-account", repr(summary)) self.assertNotIn("internal.example", repr(summary)) self.assertNotIn("secret", repr(summary)) def test_send_compares_revision_before_provider_effect(self) -> None: profile = _profile() with ( patch( "govoplan_mail.backend.capabilities._authorized_campaign_profile", return_value=profile, ), patch( "govoplan_mail.backend.capabilities.smtp_config_from_profile", side_effect=AssertionError("stale revisions must be rejected before SMTP credential decryption"), ), patch( "govoplan_mail.backend.capabilities.campaign_profile_transport_revisions", return_value={"smtp": "current", "imap": None}, ), patch("govoplan_mail.backend.capabilities.send_email_bytes") as provider_send, ): with self.assertRaisesRegex(MailProfileError, "changed after this campaign was built"): send_campaign_email_bytes( object(), # type: ignore[arg-type] tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", message_bytes=b"message", envelope_from="sender@example.test", envelope_recipients=["recipient@example.test"], from_header="sender@example.test", expected_smtp_transport_revision="stale", ) provider_send.assert_not_called() def test_stale_imap_revision_is_rejected_before_credential_decryption(self) -> None: profile = _profile() profile.imap_config = { "host": "imap.internal.example", "port": 993, "security": "tls", } with ( patch( "govoplan_mail.backend.capabilities._authorized_campaign_profile", return_value=profile, ), patch( "govoplan_mail.backend.capabilities.campaign_profile_transport_revisions", return_value={"smtp": "smtp-current", "imap": "imap-current"}, ), patch( "govoplan_mail.backend.capabilities.imap_config_from_profile", side_effect=AssertionError("stale revisions must be rejected before IMAP credential decryption"), ), patch("govoplan_mail.backend.capabilities.append_message_to_sent") as provider_append, ): with self.assertRaisesRegex(MailProfileError, "changed after this campaign was built"): append_campaign_message_to_sent( object(), # type: ignore[arg-type] tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", message_bytes=b"message", folder="Sent", expected_smtp_transport_revision="smtp-current", expected_imap_transport_revision="imap-stale", ) provider_append.assert_not_called() def test_provider_response_is_sanitized_before_campaign_evidence(self) -> None: profile = _profile() smtp = SmtpConfig(host="smtp.internal.example", port=587) provider_secret = b"smtp.internal.example says credential=provider-secret" provider_result = SimpleNamespace( envelope_recipients=["ok@example.test", "blocked@example.test"], refused_recipients={"blocked@example.test": (550, provider_secret)}, ) with ( patch( "govoplan_mail.backend.capabilities._authorized_campaign_profile", return_value=profile, ), patch("govoplan_mail.backend.capabilities.smtp_config_from_profile", return_value=smtp), patch( "govoplan_mail.backend.capabilities.imap_config_from_profile", side_effect=AssertionError("SMTP delivery must not decrypt IMAP credentials"), ), patch( "govoplan_mail.backend.capabilities.campaign_profile_transport_revisions", return_value={"smtp": "current", "imap": None}, ), patch("govoplan_mail.backend.capabilities.assert_mail_policy_allows_send"), patch("govoplan_mail.backend.capabilities.send_email_bytes", return_value=provider_result), ): result = send_campaign_email_bytes( object(), # type: ignore[arg-type] tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", message_bytes=b"message", envelope_from="sender@example.test", envelope_recipients=provider_result.envelope_recipients, from_header="sender@example.test", expected_smtp_transport_revision="current", ) evidence = json.dumps(asdict(result), sort_keys=True) self.assertEqual(result.accepted_count, 1) self.assertIn('"status_code": 550', evidence) self.assertIn('"classification": "permanent"', evidence) self.assertNotIn("smtp.internal.example", evidence) self.assertNotIn("provider-secret", evidence) def test_append_materializes_only_imap_credentials_and_preserves_unknown_outcome(self) -> None: profile = _profile() profile.imap_config = { "host": "imap.internal.example", "port": 993, "security": "tls", } imap = ImapConfig(host="imap.internal.example", port=993) with ( patch( "govoplan_mail.backend.capabilities._authorized_campaign_profile", return_value=profile, ), patch("govoplan_mail.backend.capabilities.imap_config_from_profile", return_value=imap), patch( "govoplan_mail.backend.capabilities.smtp_config_from_profile", side_effect=AssertionError("IMAP append must not decrypt SMTP credentials"), ), patch( "govoplan_mail.backend.capabilities.campaign_profile_transport_revisions", return_value={"smtp": "smtp-current", "imap": "imap-current"}, ), patch("govoplan_mail.backend.capabilities.assert_mail_policy_allows_send"), patch( "govoplan_mail.backend.capabilities.append_message_to_sent", side_effect=ImapAppendError( "imap.internal.example provider-secret", outcome_unknown=True, ), ), ): with self.assertRaises(ImapAppendError) as captured: append_campaign_message_to_sent( object(), # type: ignore[arg-type] tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", message_bytes=b"message", folder="Sent", expected_smtp_transport_revision="smtp-current", expected_imap_transport_revision="imap-current", ) self.assertTrue(captured.exception.outcome_unknown) self.assertFalse(captured.exception.temporary) self.assertIn("inspect the mailbox", str(captured.exception)) self.assertNotIn("internal.example", str(captured.exception)) self.assertNotIn("provider-secret", str(captured.exception)) def test_provider_exception_is_sanitized_and_bounded(self) -> None: profile = _profile() smtp = SmtpConfig(host="smtp.internal.example", port=587) with ( patch( "govoplan_mail.backend.capabilities._authorized_campaign_profile", return_value=profile, ), patch("govoplan_mail.backend.capabilities.smtp_config_from_profile", return_value=smtp), patch( "govoplan_mail.backend.capabilities.campaign_profile_transport_revisions", return_value={"smtp": "current", "imap": None}, ), patch("govoplan_mail.backend.capabilities.assert_mail_policy_allows_send"), patch( "govoplan_mail.backend.capabilities.send_email_bytes", side_effect=SmtpSendError( "smtp.internal.example provider-secret " + ("x" * 1_000), temporary=True, ), ), ): with self.assertRaises(SmtpSendError) as captured: send_campaign_email_bytes( object(), # type: ignore[arg-type] tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", message_bytes=b"message", envelope_from="sender@example.test", envelope_recipients=["recipient@example.test"], from_header="sender@example.test", expected_smtp_transport_revision="current", ) message = str(captured.exception) self.assertEqual(message, "Mail delivery failed temporarily.") self.assertNotIn("internal.example", message) self.assertLessEqual(len(message), 80) self.assertIsNone(captured.exception.__cause__) def test_unclassified_provider_exception_is_outcome_unknown(self) -> None: profile = _profile() smtp = SmtpConfig(host="smtp.internal.example", port=587) with ( patch( "govoplan_mail.backend.capabilities._authorized_campaign_profile", return_value=profile, ), patch("govoplan_mail.backend.capabilities.smtp_config_from_profile", return_value=smtp), patch( "govoplan_mail.backend.capabilities.campaign_profile_transport_revisions", return_value={"smtp": "current", "imap": None}, ), patch("govoplan_mail.backend.capabilities.assert_mail_policy_allows_send"), patch( "govoplan_mail.backend.capabilities.send_email_bytes", side_effect=RuntimeError("smtp.internal.example provider-secret"), ), ): with self.assertRaises(SmtpSendError) as captured: send_campaign_email_bytes( object(), # type: ignore[arg-type] tenant_id="tenant-1", campaign_id="campaign-1", profile_id="profile-1", message_bytes=b"message", envelope_from="sender@example.test", envelope_recipients=["recipient@example.test"], from_header="sender@example.test", expected_smtp_transport_revision="current", ) self.assertTrue(captured.exception.outcome_unknown) self.assertNotIn("internal.example", str(captured.exception)) self.assertIsNone(captured.exception.__cause__) def test_capability_does_not_export_raw_profile_or_transport_helpers(self) -> None: capability = MailCampaignCapability() for name in ( "smtp_config_from_profile", "imap_config_from_profile", "send_email_bytes", "send_email_message", "materialize_campaign_mail_profile_config", ): self.assertFalse(hasattr(capability, name), name) if __name__ == "__main__": unittest.main()