<filename>tests/test_mailgun_inbound.py
import json
from datetime import datetime
from textwrap import dedent

import six
from django.test import override_settings, tag
from django.utils.timezone import utc
from mock import ANY

from anymail.exceptions import AnymailConfigurationError
from anymail.inbound import AnymailInboundMessage
from anymail.signals import AnymailInboundEvent
from anymail.webhooks.mailgun import MailgunInboundWebhookView

from .test_mailgun_webhooks import (
    TEST_API_KEY, mailgun_sign_payload,
    mailgun_sign_legacy_payload, querydict_to_postdict)
from .utils import sample_image_content, sample_email_content
from .webhook_cases import WebhookTestCase


@tag('mailgun')
@override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY)
class MailgunInboundTestCase(WebhookTestCase):
    def test_inbound_basics(self):
        raw_event = mailgun_sign_legacy_payload({
            'token': '06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0',
            'timestamp': '1461261330',
            'recipient': '<EMAIL>',
            'sender': '<EMAIL>',
            'message-headers': json.dumps([
                ["X-Mailgun-Spam-Rules", "DKIM_SIGNED, DKIM_VALID, DKIM_VALID_AU, ..."],
                ["X-Mailgun-Dkim-Check-Result", "Pass"],
                ["X-Mailgun-Spf", "Pass"],
                ["X-Mailgun-Sscore", "1.7"],
                ["X-Mailgun-Sflag", "No"],
                ["X-Mailgun-Incoming", "Yes"],
                ["X-Envelope-From", "<<EMAIL>>"],
                ["Received", "from mail.example.org by mxa.mailgun.org ..."],
                ["Received", "by mail.example.org for <<EMAIL>> ..."],
                ["Dkim-Signature", "v=1; a=rsa-sha256; c=relaxed/relaxed; d=example.org; ..."],
                ["Mime-Version", "1.0"],
                ["Received", "by 10.10.1.71 with HTTP; Wed, 11 Oct 2017 18:31:04 -0700 (PDT)"],
                ["From", "\"Displayed From\" <<EMAIL>>"],
                ["Date", "Wed, 11 Oct 2017 18:31:04 -0700"],
                ["Message-Id", "<<EMAIL>>"],
                ["Subject", "Test subject"],
                ["To", "\"Test Inbound\" <<EMAIL>>, <EMAIL>"],
                ["Cc", "<EMAIL>"],
                ["Content-Type", "multipart/mixed; boundary=\"089e0825ccf874a0bb055b4f7e23\""],
            ]),
            'body-plain': 'Test body plain',
            'body-html': '<div>Test body html</div>',
            'stripped-html': 'stripped html body',
            'stripped-text': 'stripped plaintext body',
        })
        response = self.client.post('/anymail/mailgun/inbound/', data=raw_event)
        self.assertEqual(response.status_code, 200)
        kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=MailgunInboundWebhookView,
                                                      event=ANY, esp_name='Mailgun')
        # AnymailInboundEvent
        event = kwargs['event']
        self.assertIsInstance(event, AnymailInboundEvent)
        self.assertEqual(event.event_type, 'inbound')
        self.assertEqual(event.timestamp, datetime(2016, 4, 21, 17, 55, 30, tzinfo=utc))
        self.assertEqual(event.event_id, "06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0")
        self.assertIsInstance(event.message, AnymailInboundMessage)
        self.assertEqual(querydict_to_postdict(event.esp_event.POST), raw_event)

        # AnymailInboundMessage - convenience properties
        message = event.message

        self.assertEqual(message.from_email.display_name, 'Displayed From')
        self.assertEqual(message.from_email.addr_spec, '<EMAIL>')
        self.assertEqual([str(e) for e in message.to],
                         ['Test Inbound <<EMAIL>>', '<EMAIL>'])
        self.assertEqual([str(e) for e in message.cc],
                         ['<EMAIL>'])
        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, '<div>Test body html</div>')

        self.assertEqual(message.envelope_sender, '<EMAIL>')
        self.assertEqual(message.envelope_recipient, '<EMAIL>')
        self.assertEqual(message.stripped_text, 'stripped plaintext body')
        self.assertEqual(message.stripped_html, 'stripped html body')
        self.assertIs(message.spam_detected, False)
        self.assertEqual(message.spam_score, 1.7)

        # AnymailInboundMessage - other headers
        self.assertEqual(message['Message-ID'], "<<EMAIL>>")
        self.assertEqual(message.get_all('Received'), [
            "from mail.example.org by mxa.mailgun.org ...",
            "by mail.example.org for <<EMAIL>> ...",
            "by 10.10.1.71 with HTTP; Wed, 11 Oct 2017 18:31:04 -0700 (PDT)",
        ])

    def test_attachments(self):
        att1 = six.BytesIO('test attachment'.encode('utf-8'))
        att1.name = 'test.txt'
        image_content = sample_image_content()
        att2 = six.BytesIO(image_content)
        att2.name = 'image.png'
        email_content = sample_email_content()
        att3 = six.BytesIO(email_content)
        att3.content_type = 'message/rfc822; charset="us-ascii"'
        raw_event = mailgun_sign_legacy_payload({
            'message-headers': '[]',
            'attachment-count': '3',
            'content-id-map': """{"<abc123>": "attachment-2"}""",
            'attachment-1': att1,
            'attachment-2': att2,  # inline
            'attachment-3': att3,
        })

        response = self.client.post('/anymail/mailgun/inbound/', data=raw_event)
        self.assertEqual(response.status_code, 200)
        kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=MailgunInboundWebhookView,
                                                      event=ANY, esp_name='Mailgun')
        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(), u'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)

    def test_inbound_mime(self):
        # Mailgun provides the full, raw MIME message if the webhook url ends in 'mime'
        raw_event = mailgun_sign_legacy_payload({
            'token': '06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0',
            'timestamp': '1461261330',
            'recipient': '<EMAIL>',
            'sender': '<EMAIL>',
            'body-mime': dedent("""\
                From: A tester <<EMAIL>>
                Date: Thu, 12 Oct 2017 18:03:30 -0700
                Message-ID: <<EMAIL>>
                Subject: Raw MIME test
                To: <EMAIL>
                MIME-Version: 1.0
                Content-Type: multipart/alternative; boundary="94eb2c05e174adb140055b6339c5"

                --94eb2c05e174adb140055b6339c5
                Content-Type: text/plain; charset="UTF-8"
                Content-Transfer-Encoding: quoted-printable

                It's a body=E2=80=A6

                --94eb2c05e174adb140055b6339c5
                Content-Type: text/html; charset="UTF-8"
                Content-Transfer-Encoding: quoted-printable

                <div dir=3D"ltr">It's a body=E2=80=A6</div>

                --94eb2c05e174adb140055b6339c5--
                """),
        })

        response = self.client.post('/anymail/mailgun/inbound_mime/', data=raw_event)
        self.assertEqual(response.status_code, 200)
        kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=MailgunInboundWebhookView,
                                                      event=ANY, esp_name='Mailgun')
        event = kwargs['event']
        message = event.message
        self.assertEqual(message.envelope_sender, '<EMAIL>')
        self.assertEqual(message.envelope_recipient, '<EMAIL>')
        self.assertEqual(message.subject, 'Raw MIME test')
        self.assertEqual(message.text, u"It's a body\N{HORIZONTAL ELLIPSIS}\n")
        self.assertEqual(message.html, u"""<div dir="ltr">It's a body\N{HORIZONTAL ELLIPSIS}</div>\n""")

    def test_misconfigured_tracking(self):
        raw_event = mailgun_sign_payload({
            "event-data": {
                "event": "clicked",
                "timestamp": 1534109600.089676,
                "recipient": "<EMAIL>",
                "url": "https://example.com/test"
            }
        })
        with self.assertRaisesMessage(
            AnymailConfigurationError,
            "You seem to have set Mailgun's *clicked tracking* webhook"
            " to Anymail's Mailgun *inbound* webhook URL."
        ):
            self.client.post('/anymail/mailgun/inbound/',
                             data=json.dumps(raw_event), content_type='application/json')

    def test_misconfigured_tracking_legacy(self):
        raw_event = mailgun_sign_legacy_payload({
            'domain': 'example.com',
            'message-headers': '[]',
            'recipient': '<EMAIL>',
            'event': 'delivered',
        })
        with self.assertRaisesMessage(
            AnymailConfigurationError,
            "You seem to have set Mailgun's *delivered tracking* webhook"
            " to Anymail's Mailgun *inbound* webhook URL."
        ):
            self.client.post('/anymail/mailgun/inbound/', data=raw_event)

<filename>tests/test_mailgun_inbound.py
import json
from datetime import datetime
from textwrap import dedent

import six
from django.test import override_settings, tag
from django.utils.timezone import utc
from mock import ANY

from anymail.exceptions import AnymailConfigurationError
from anymail.inbound import AnymailInboundMessage
from anymail.signals import AnymailInboundEvent
from anymail.webhooks.mailgun import MailgunInboundWebhookView

from .test_mailgun_webhooks import (
    TEST_API_KEY, mailgun_sign_payload,
    mailgun_sign_legacy_payload, querydict_to_postdict)
from .utils import sample_image_content, sample_email_content
from .webhook_cases import WebhookTestCase


@tag('mailgun')
@override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY)
class MailgunInboundTestCase(WebhookTestCase):
    def test_inbound_basics(self):
        raw_event = mailgun_sign_legacy_payload({
            'token': '06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0',
            'timestamp': '1461261330',
            'recipient': '<EMAIL>',
            'sender': '<EMAIL>',
            'message-headers': json.dumps([
                ["X-Mailgun-Spam-Rules", "DKIM_SIGNED, DKIM_VALID, DKIM_VALID_AU, ..."],
                ["X-Mailgun-Dkim-Check-Result", "Pass"],
                ["X-Mailgun-Spf", "Pass"],
                ["X-Mailgun-Sscore", "1.7"],
                ["X-Mailgun-Sflag", "No"],
                ["X-Mailgun-Incoming", "Yes"],
                ["X-Envelope-From", "<<EMAIL>>"],
                ["Received", "from mail.example.org by mxa.mailgun.org ..."],
                ["Received", "by mail.example.org for <<EMAIL>> ..."],
                ["Dkim-Signature", "v=1; a=rsa-sha256; c=relaxed/relaxed; d=example.org; ..."],
                ["Mime-Version", "1.0"],
                ["Received", "by 10.10.1.71 with HTTP; Wed, 11 Oct 2017 18:31:04 -0700 (PDT)"],
                ["From", "\"Displayed From\" <<EMAIL>>"],
                ["Date", "Wed, 11 Oct 2017 18:31:04 -0700"],
                ["Message-Id", "<<EMAIL>>"],
                ["Subject", "Test subject"],
                ["To", "\"Test Inbound\" <<EMAIL>>, <EMAIL>"],
                ["Cc", "<EMAIL>"],
                ["Content-Type", "multipart/mixed; boundary=\"089e0825ccf874a0bb055b4f7e23\""],
            ]),
            'body-plain': 'Test body plain',
            'body-html': '<div>Test body html</div>',
            'stripped-html': 'stripped html body',
            'stripped-text': 'stripped plaintext body',
        })
        response = self.client.post('/anymail/mailgun/inbound/', data=raw_event)
        self.assertEqual(response.status_code, 200)
        kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=MailgunInboundWebhookView,
                                                      event=ANY, esp_name='Mailgun')
        # AnymailInboundEvent
        event = kwargs['event']
        self.assertIsInstance(event, AnymailInboundEvent)
        self.assertEqual(event.event_type, 'inbound')
        self.assertEqual(event.timestamp, datetime(2016, 4, 21, 17, 55, 30, tzinfo=utc))
        self.assertEqual(event.event_id, "06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0")
        self.assertIsInstance(event.message, AnymailInboundMessage)
        self.assertEqual(querydict_to_postdict(event.esp_event.POST), raw_event)

        # AnymailInboundMessage - convenience properties
        message = event.message

        self.assertEqual(message.from_email.display_name, 'Displayed From')
        self.assertEqual(message.from_email.addr_spec, '<EMAIL>')
        self.assertEqual([str(e) for e in message.to],
                         ['Test Inbound <<EMAIL>>', '<EMAIL>'])
        self.assertEqual([str(e) for e in message.cc],
                         ['<EMAIL>'])
        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, '<div>Test body html</div>')

        self.assertEqual(message.envelope_sender, '<EMAIL>')
        self.assertEqual(message.envelope_recipient, '<EMAIL>')
        self.assertEqual(message.stripped_text, 'stripped plaintext body')
        self.assertEqual(message.stripped_html, 'stripped html body')
        self.assertIs(message.spam_detected, False)
        self.assertEqual(message.spam_score, 1.7)

        # AnymailInboundMessage - other headers
        self.assertEqual(message['Message-ID'], "<<EMAIL>>")
        self.assertEqual(message.get_all('Received'), [
            "from mail.example.org by mxa.mailgun.org ...",
            "by mail.example.org for <<EMAIL>> ...",
            "by 10.10.1.71 with HTTP; Wed, 11 Oct 2017 18:31:04 -0700 (PDT)",
        ])

    def test_attachments(self):
        att1 = six.BytesIO('test attachment'.encode('utf-8'))
        att1.name = 'test.txt'
        image_content = sample_image_content()
        att2 = six.BytesIO(image_content)
        att2.name = 'image.png'
        email_content = sample_email_content()
        att3 = six.BytesIO(email_content)
        att3.content_type = 'message/rfc822; charset="us-ascii"'
        raw_event = mailgun_sign_legacy_payload({
            'message-headers': '[]',
            'attachment-count': '3',
            'content-id-map': """{"<abc123>": "attachment-2"}""",
            'attachment-1': att1,
            'attachment-2': att2,  # inline
            'attachment-3': att3,
        })

        response = self.client.post('/anymail/mailgun/inbound/', data=raw_event)
        self.assertEqual(response.status_code, 200)
        kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=MailgunInboundWebhookView,
                                                      event=ANY, esp_name='Mailgun')
        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(), u'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)

    def test_inbound_mime(self):
        # Mailgun provides the full, raw MIME message if the webhook url ends in 'mime'
        raw_event = mailgun_sign_legacy_payload({
            'token': '06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0',
            'timestamp': '1461261330',
            'recipient': '<EMAIL>',
            'sender': '<EMAIL>',
            'body-mime': dedent("""\
                From: A tester <<EMAIL>>
                Date: Thu, 12 Oct 2017 18:03:30 -0700
                Message-ID: <<EMAIL>>
                Subject: Raw MIME test
                To: <EMAIL>
                MIME-Version: 1.0
                Content-Type: multipart/alternative; boundary="94eb2c05e174adb140055b6339c5"

                --94eb2c05e174adb140055b6339c5
                Content-Type: text/plain; charset="UTF-8"
                Content-Transfer-Encoding: quoted-printable

                It's a body=E2=80=A6

                --94eb2c05e174adb140055b6339c5
                Content-Type: text/html; charset="UTF-8"
                Content-Transfer-Encoding: quoted-printable

                <div dir=3D"ltr">It's a body=E2=80=A6</div>

                --94eb2c05e174adb140055b6339c5--
                """),
        })

        response = self.client.post('/anymail/mailgun/inbound_mime/', data=raw_event)
        self.assertEqual(response.status_code, 200)
        kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=MailgunInboundWebhookView,
                                                      event=ANY, esp_name='Mailgun')
        event = kwargs['event']
        message = event.message
        self.assertEqual(message.envelope_sender, '<EMAIL>')
        self.assertEqual(message.envelope_recipient, '<EMAIL>')
        self.assertEqual(message.subject, 'Raw MIME test')
        self.assertEqual(message.text, u"It's a body\N{HORIZONTAL ELLIPSIS}\n")
        self.assertEqual(message.html, u"""<div dir="ltr">It's a body\N{HORIZONTAL ELLIPSIS}</div>\n""")

    def test_misconfigured_tracking(self):
        raw_event = mailgun_sign_payload({
            "event-data": {
                "event": "clicked",
                "timestamp": 1534109600.089676,
                "recipient": "<EMAIL>",
                "url": "https://example.com/test"
            }
        })
        with self.assertRaisesMessage(
            AnymailConfigurationError,
            "You seem to have set Mailgun's *clicked tracking* webhook"
            " to Anymail's Mailgun *inbound* webhook URL."
        ):
            self.client.post('/anymail/mailgun/inbound/',
                             data=json.dumps(raw_event), content_type='application/json')

    def test_misconfigured_tracking_legacy(self):
        raw_event = mailgun_sign_legacy_payload({
            'domain': 'example.com',
            'message-headers': '[]',
            'recipient': '<EMAIL>',
            'event': 'delivered',
        })
        with self.assertRaisesMessage(
            AnymailConfigurationError,
            "You seem to have set Mailgun's *delivered tracking* webhook"
            " to Anymail's Mailgun *inbound* webhook URL."
        ):
            self.client.post('/anymail/mailgun/inbound/', data=raw_event)

# AnymailInboundEvent # AnymailInboundMessage - convenience properties # AnymailInboundMessage - other headers {"<abc123>": "attachment-2"} # inline # AnymailInboundMessage convenience accessor # Mailgun provides the full, raw MIME message if the webhook url ends in 'mime' \                 From: A tester <<EMAIL>>                 Date: Thu, 12 Oct 2017 18:03:30 -0700                 Message-ID: <<EMAIL>>                 Subject: Raw MIME test                 To: <EMAIL>                 MIME-Version: 1.0                 Content-Type: multipart/alternative; boundary="94eb2c05e174adb140055b6339c5"                  --94eb2c05e174adb140055b6339c5                 Content-Type: text/plain; charset="UTF-8"                 Content-Transfer-Encoding: quoted-printable                  It's a body=E2=80=A6                  --94eb2c05e174adb140055b6339c5                 Content-Type: text/html; charset="UTF-8"                 Content-Transfer-Encoding: quoted-printable                  <div dir=3D"ltr">It's a body=E2=80=A6</div>                  --94eb2c05e174adb140055b6339c5-- <div dir="ltr">It's a body\N{HORIZONTAL ELLIPSIS}</div>\n