92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from email import message_from_bytes
|
|
from unittest.mock import ANY, patch
|
|
|
|
from govoplan_core.core.mail import NotificationMailDeliveryRequest
|
|
from govoplan_mail.backend.capabilities import MailNotificationDeliveryCapability
|
|
|
|
|
|
class MailNotificationDeliveryCapabilityTests(unittest.TestCase):
|
|
def test_missing_policy_selection_pauses_without_transport_effect(self) -> None:
|
|
capability = MailNotificationDeliveryCapability()
|
|
|
|
with patch(
|
|
"govoplan_mail.backend.delivery_outbox.submit_delivery_command"
|
|
) as submit:
|
|
result = capability.submit_notification_mail(
|
|
object(),
|
|
NotificationMailDeliveryRequest(
|
|
tenant_id="tenant-1",
|
|
notification_id="notification-1",
|
|
recipient="person@example.test",
|
|
subject="Subject",
|
|
body_text="Body",
|
|
),
|
|
)
|
|
|
|
self.assertEqual(result["status"], "paused")
|
|
submit.assert_not_called()
|
|
|
|
def test_notification_is_submitted_to_durable_mail_outbox(self) -> None:
|
|
capability = MailNotificationDeliveryCapability()
|
|
transport = {
|
|
"smtp_available": True,
|
|
"smtp_transport_revision": "revision-1",
|
|
"smtp_server_id": "server-1",
|
|
"smtp_credential_id": "credential-1",
|
|
}
|
|
command = {
|
|
"id": "command-1",
|
|
"status": "pending",
|
|
"duplicate": False,
|
|
}
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_mail.backend.capabilities.campaign_profile_delivery_summary",
|
|
return_value=transport,
|
|
) as summary,
|
|
patch(
|
|
"govoplan_mail.backend.delivery_outbox.submit_delivery_command",
|
|
return_value=command,
|
|
) as submit,
|
|
):
|
|
result = capability.submit_notification_mail(
|
|
object(),
|
|
NotificationMailDeliveryRequest(
|
|
tenant_id="tenant-1",
|
|
notification_id="notification-1",
|
|
recipient="person@example.test",
|
|
subject="Subject",
|
|
body_text="Body",
|
|
body_html="<p>Body</p>",
|
|
mail_profile_id="profile-1",
|
|
from_address="notifications@example.test",
|
|
),
|
|
)
|
|
|
|
self.assertEqual(result["status"], "accepted")
|
|
self.assertEqual(result["external_message_id"], "command-1")
|
|
summary.assert_called_once_with(
|
|
ANY,
|
|
tenant_id="tenant-1",
|
|
profile_id="profile-1",
|
|
smtp_server_id=None,
|
|
smtp_credential_id=None,
|
|
)
|
|
payload = submit.call_args.kwargs
|
|
self.assertEqual(payload["idempotency_key"], "notification:notification-1")
|
|
self.assertEqual(payload["profile_id"], "profile-1")
|
|
self.assertEqual(payload["smtp_server_id"], "server-1")
|
|
self.assertEqual(payload["smtp_credential_id"], "credential-1")
|
|
message = message_from_bytes(payload["message_bytes"])
|
|
self.assertEqual(message["From"], "notifications@example.test")
|
|
self.assertEqual(message["To"], "person@example.test")
|
|
self.assertEqual(message["Subject"], "Subject")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|