import json from base64 import b64encode from unittest.mock import ANY from django.test import tag from anymail.inbound import AnymailInboundMessage from anymail.signals import AnymailInboundEvent from anymail.webhooks.mailjet import MailjetInboundWebhookView from .utils import sample_email_content, sample_image_content from .webhook_cases import WebhookTestCase @tag("mailjet") class MailjetInboundTestCase(WebhookTestCase): def test_inbound_basics(self): raw_event = { "Sender": "envelope-from@example.org", "Recipient": "test@inbound.example.com", # this is just the Date header from the sender, parsed to UTC: "Date": "20171012T013104", "From": '"Displayed From" ', "Subject": "Test subject", "Headers": { "Return-Path": [ "" ], "Received": [ "from mail.example.org by parse.mailjet.com ..." "by mail.example.org for ...", "by 10.10.1.71 with HTTP; Wed, 11 Oct 2017 18:31:04 -0700 (PDT)", ], "MIME-Version": ["1.0"], "From": '"Displayed From" ', "Date": "Wed, 11 Oct 2017 18:31:04 -0700", "Message-ID": "", "Subject": "Test subject", "To": "Test Inbound , other@example.com", "Cc": "cc@example.com", "Reply-To": "from+test@milter.example.org", "Content-Type": ['multipart/alternative; boundary="boundary0"'], }, "Parts": [ { "Headers": {"Content-Type": ['text/plain; charset="UTF-8"']}, "ContentRef": "Text-part", }, { "Headers": { "Content-Type": ['text/html; charset="UTF-8"'], "Content-Transfer-Encoding": ["quoted-printable"], }, "ContentRef": "Html-part", }, ], "Text-part": "Test body plain", "Html-part": "
Test body html
", "SpamAssassinScore": "1.7", } response = self.client.post( "/anymail/mailjet/inbound/", content_type="application/json", data=json.dumps(raw_event), ) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.inbound_handler, sender=MailjetInboundWebhookView, event=ANY, esp_name="Mailjet", ) # AnymailInboundEvent event = kwargs["event"] self.assertIsInstance(event, AnymailInboundEvent) self.assertEqual(event.event_type, "inbound") # Mailjet doesn't provide inbound event timestamp: self.assertIsNone(event.timestamp) # Mailjet doesn't provide inbound event id: self.assertIsNone(event.event_id) self.assertIsInstance(event.message, AnymailInboundMessage) self.assertEqual(event.esp_event, raw_event) # AnymailInboundMessage - convenience properties message = event.message self.assertEqual(message.from_email.display_name, "Displayed From") self.assertEqual(message.from_email.addr_spec, "from+test@example.org") self.assertEqual( [str(e) for e in message.to], ["Test Inbound ", "other@example.com"], ) self.assertEqual([str(e) for e in message.cc], ["cc@example.com"]) self.assertEqual(message.subject, "Test subject") self.assertEqual(message.date.isoformat(" "), "2017-10-11 18:31:04-07:00") self.assertEqual(message.text, "Test body plain") self.assertEqual(message.html, "
Test body html
") self.assertEqual(message.envelope_sender, "envelope-from@example.org") self.assertEqual(message.envelope_recipient, "test@inbound.example.com") # Mailjet doesn't provide stripped plaintext body: self.assertIsNone(message.stripped_text) # Mailjet doesn't provide stripped html: self.assertIsNone(message.stripped_html) # Mailjet doesn't provide spam boolean: self.assertIsNone(message.spam_detected) self.assertEqual(message.spam_score, 1.7) # AnymailInboundMessage - other headers self.assertEqual(message["Message-ID"], "") self.assertEqual(message["Reply-To"], "from+test@milter.example.org") self.assertEqual( message.get_all("Received"), [ "from mail.example.org by parse.mailjet.com ..." "by mail.example.org for ...", "by 10.10.1.71 with HTTP; Wed, 11 Oct 2017 18:31:04 -0700 (PDT)", ], ) def test_attachments(self): image_content = sample_image_content() email_content = sample_email_content() raw_event = { "Headers": { "MIME-Version": ["1.0"], "Content-Type": ['multipart/mixed; boundary="boundary0"'], }, "Parts": [ { "Headers": { "Content-Type": ['multipart/related; boundary="boundary1"'] } }, { "Headers": { "Content-Type": ['multipart/alternative; boundary="boundary2"'] } }, { "Headers": {"Content-Type": ['text/plain; charset="UTF-8"']}, "ContentRef": "Text-part", }, { "Headers": { "Content-Type": ['text/html; charset="UTF-8"'], "Content-Transfer-Encoding": ["quoted-printable"], }, "ContentRef": "Html-part", }, { "Headers": { "Content-Type": ["text/plain"], "Content-Disposition": ['attachment; filename="test.txt"'], "Content-Transfer-Encoding": ["quoted-printable"], }, "ContentRef": "Attachment1", }, { "Headers": { "Content-Type": ['image/png; name="image.png"'], "Content-Disposition": ['inline; filename="image.png"'], "Content-Transfer-Encoding": ["base64"], "Content-ID": [""], }, "ContentRef": "InlineAttachment1", }, { "Headers": { "Content-Type": ['message/rfc822; charset="US-ASCII"'], "Content-Disposition": ["attachment"], }, "ContentRef": "Attachment2", }, ], "Text-part": "Test body plain", "Html-part": "
Test body html
", "InlineAttachment1": b64encode(image_content).decode("ascii"), "Attachment1": b64encode("test attachment".encode("utf-8")).decode("ascii"), "Attachment2": b64encode(email_content).decode("ascii"), } response = self.client.post( "/anymail/mailjet/inbound/", content_type="application/json", data=json.dumps(raw_event), ) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.inbound_handler, sender=MailjetInboundWebhookView, event=ANY, esp_name="Mailjet", ) event = kwargs["event"] message = event.message attachments = message.attachments # AnymailInboundMessage convenience accessor self.assertEqual(len(attachments), 2) self.assertEqual(attachments[0].get_filename(), "test.txt") self.assertEqual(attachments[0].get_content_type(), "text/plain") self.assertEqual(attachments[0].get_content_text(), "test attachment") self.assertEqual(attachments[1].get_content_type(), "message/rfc822") self.assertEqualIgnoringHeaderFolding( attachments[1].get_content_bytes(), email_content ) inlines = message.inline_attachments self.assertEqual(len(inlines), 1) inline = inlines["abc123"] self.assertEqual(inline.get_filename(), "image.png") self.assertEqual(inline.get_content_type(), "image/png") self.assertEqual(inline.get_content_bytes(), image_content)