323 lines
14 KiB
Python
323 lines
14 KiB
Python
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()
|