mail-addons/email_headers/tests/test_email.py

import odoo
from odoo.tests import TransactionCase
from email.message import EmailMessage
from ..models.mail import encode_msg_id
from ..models.mail import encode_msg_id_legacy
from ..models.mail import MESSAGE_PREFIX
import mock
from odoo import SUPERUSER_ID
from odoo.addons.base.models.ir_mail_server import IrMailServer
from ..models.mail import random_string


@odoo.tests.tagged('post_install', '-at_install')
class TestEmail(TransactionCase):

    def setUp(self):
        super(TestEmail, self).setUp()

        self.partner = self.env['res.partner'].create({'name': 'Test Dude'})
        self.partner2 = self.env['res.partner'].create({'name': 'Dudette'})
        self.demo_user = self.env.ref('base.user_demo')
        self.subtype_comment = self.env.ref('mail.mt_comment')
        self.subtype_note = self.env.ref('mail.mt_note')

        self.MailMessage = self.env['mail.message']
        self.ConfigParam = self.env['ir.config_parameter']

        # Create server configuration
        self.outgoing_server = self.env['ir.mail_server'].create({
            'name': 'Outgoing SMTP Server for Unit Tests',
            'sequence': 1,
            'smtp_host': 'localhost',
            'smtp_port': '9999',
            'smtp_encryption': 'none',
            'smtp_user': 'doesnt',
            'smtp_pass': 'exist',
            'reply_to_method': 'msg_id',
            'force_email_reply_to_domain': 'example.com',
            'force_email_from': 'odoo@example.com',
        })

    @staticmethod
    def create_email_message():
        message = EmailMessage()
        message['Content-Type'] = 'multipart/mixed; boundary="===============2590914155756834027=="'
        message['MIME-Version'] = '1.0'
        message['Message-Id'] = '<CAD-eYi=a264_3DcrYSDU5yc_fwYoHonZ3H+{}@mail.gmail.com>'.format(random_string(6))
        message['Subject'] = '1'
        message['From'] = 'Miku Laitinen <miku@avoin.systems>'
        message['Reply-To'] = 'YourCompany Eurooppa <sales@avoin.onmicrosoft.com>'
        message['To'] = '"Erik N. French" <ErikNFrench@armyspy.com>'
        message['Date'] = 'Mon, 06 May 2019 14:16:38 -0000'
        return message

    def test_reply_to_method_msg_id(self):

        # Make administrator follow the partner
        self.partner.message_subscribe([self.env.user.partner_id.id])

        # Send a message to the followers of the partner
        thread_msg = self.partner.with_user(self.demo_user).message_post(
            body='dummy message.',
            message_type='comment',
            subtype='mail.mt_comment'
        )

        # Make sure the message headers look right.. or not
        # mail_msg = thread_msg.notification_ids[0]

        # Get the encoded message address
        encoded_msg_id = encode_msg_id(thread_msg.id, self.env)

        # Try to read an incoming email
        message = self.create_email_message()
        del message['To']
        message['To'] = '"Erik N. French" <ErikNFrench+{}{}@armyspy.com>'.format(MESSAGE_PREFIX, encoded_msg_id)

        thread_id = self.env['mail.thread'].message_process(model=False, message=message.as_string())
        self.assertEqual(thread_msg.res_id, thread_id,
                         "The incoming email wasn't connected to the correct thread")

        # Make sure the message is a comment
        incoming_msg1 = self.MailMessage.search([('message_id', '=', message['Message-Id'])])
        self.assertEqual(
            incoming_msg1.message_type,
            'email',
            "The incoming message was created as a type {} instead of a email.".format(incoming_msg1.message_type)
        )
        self.assertEqual(
            incoming_msg1.subtype_id,
            self.subtype_comment,
            "The incoming message was created as a subtype {} instead of a comment.".format(incoming_msg1.subtype_id)
        )

        # Try to read another incoming email
        message = self.create_email_message()
        del message['To']
        message['To'] = '"Erik N. French" <ErikNFrench+{}HURDURLUR@armyspy.com>'.format(MESSAGE_PREFIX)
        message['In-Reply-To'] = thread_msg.message_id

        thread_id = self.env['mail.thread'].message_process(model=False, message=message.as_string())
        self.assertEqual(thread_msg.res_id, thread_id,
                         "The incoming email wasn't connected to the correct thread")

        # Make sure the message is a comment
        incoming_msg2 = self.MailMessage.search([('message_id', '=', message['Message-Id'])])
        self.assertEqual(
            incoming_msg2.message_type,
            'email',
            "The incoming message was created as a type {} instead of a email.".format(incoming_msg2.message_type)
        )
        self.assertEqual(
            incoming_msg2.subtype_id,
            self.subtype_comment,
            "The incoming message was created as a subtype {} instead of a comment.".format(incoming_msg2.subtype_id)
        )

    def test_reply_to_method_msg_id_priority(self):
        """
        In this test we will inject the wrong Message-Id to the incoming
        email messages References-header and see if Odoo will prioritize
        the custom Reply-To address over the References-header. It should.
        :return:
        """

        # Make administrator follow the partner
        self.partner.message_subscribe([self.env.user.partner_id.id])

        # Send a message to the followers of the partner
        thread_msg = self.partner.with_user(self.demo_user).message_post(
            body='dummy message X.',
            message_type='comment',
            subtype='mail.mt_comment'
        )

        # Get the encoded message address
        encoded_msg_id = encode_msg_id(thread_msg.id, self.env)

        # Send another message to the followers of the partner
        thread_msg2 = self.partner2.with_user(self.demo_user).message_post(
            body='dummy message X.',
            message_type='comment',
            subtype='mail.mt_comment'
        )

        # Try to read an incoming email
        message = self.create_email_message()
        del message['To']
        del message['References']
        message['To'] = '"Erik N. French" <ErikNFrench+{}{}@armyspy.com>'.format(MESSAGE_PREFIX, encoded_msg_id)

        # Inject the wrong References
        message['References'] = thread_msg2.message_id

        thread_id = self.env['mail.thread'].message_process(model=False, message=message.as_string())
        self.assertEqual(thread_msg.res_id, thread_id,
                         "The incoming email wasn't connected to the correct thread")

    def test_reply_to_method_msg_id_notification(self):

        # Make administrator follow the partner
        self.partner2.message_subscribe([self.env.user.partner_id.id])

        # Send a message to the followers of the partner
        thread_msg = self.partner2.with_user(self.demo_user).message_post(
            body='dummy message 2.',
            message_type='comment',
            subtype='mail.mt_note'
        )

        # Get the encoded message address
        encoded_msg_id = encode_msg_id(thread_msg.id, self.env)

        # Try to read an incoming email
        message = self.create_email_message()
        del message['To']
        message['To'] = '"Erik N. French" <ErikNFrench+{}{}@armyspy.com>'.format(MESSAGE_PREFIX, encoded_msg_id)

        thread_id = self.env['mail.thread'].message_process(model=False, message=message.as_string())
        self.assertEqual(thread_msg.res_id, thread_id,
                         "The incoming email wasn't connected to the correct thread")

        # Make sure the message is a note
        incoming_msg1 = self.MailMessage.search([('message_id', '=', message['Message-Id'])])
        self.assertEqual(
            incoming_msg1.message_type,
            'email',
            "The incoming message was created as a type {} instead of a email.".format(incoming_msg1.message_type)
        )
        self.assertEqual(
            incoming_msg1.subtype_id,
            self.subtype_note,
            "The incoming message was created as a subtype {} instead of a note.".format(incoming_msg1.subtype_id)
        )

    def test_reply_to_method_msg_id_legacy(self):
        # REMOVE this test when porting to Odoo 14

        # Make administrator follow the partner
        self.partner2.message_subscribe([self.env.user.partner_id.id])

        # Send a message to the followers of the partner
        thread_msg = self.partner2.with_user(self.demo_user).message_post(
            body='dummy message 2.',
            message_type='comment',
            subtype='mail.mt_note'
        )

        # Get the encoded message address
        encoded_msg_id = encode_msg_id_legacy(thread_msg.id, self.env)

        # Try to read an incoming email
        message = self.create_email_message()
        del message['To']
        message['To'] = '"Erik N. French" <ErikNFrench+{}{}@armyspy.com>'.format(MESSAGE_PREFIX, encoded_msg_id)

        thread_id = self.env['mail.thread'].message_process(model=False, message=message.as_string())
        self.assertEqual(thread_msg.res_id, thread_id,
                         "The incoming email wasn't connected to the correct thread")

    def test_reply_to_method_msg_id_lowercase(self):
        # Make administrator follow the partner
        self.partner2.message_subscribe([self.env.user.partner_id.id])

        # Send a message to the followers of the partner
        thread_msg = self.partner2.with_user(self.demo_user).message_post(
            body='dummy message 2.',
            message_type='comment',
            subtype='mail.mt_note'
        )

        # Get the encoded message address
        encoded_msg_id = encode_msg_id(thread_msg.id, self.env).lower()

        # Try to read an incoming email
        message = self.create_email_message()
        del message['To']
        message['To'] = '"Erik N. French" <ErikNFrench+{}{}@armyspy.com>'.format(MESSAGE_PREFIX, encoded_msg_id)

        thread_id = self.env['mail.thread'].message_process(model=False, message=message.as_string())
        self.assertEqual(thread_msg.res_id, thread_id,
                         "The incoming email wasn't connected to the correct thread")

    def test_outgoing_msg_id(self):
        # Make administrator follow the partner
        self.partner2.message_subscribe([SUPERUSER_ID])

        with mock.patch.object(IrMailServer, 'send_email') as send_email:
            # Send a message to the followers of the partner
            thread_msg = self.partner2.with_user(self.demo_user).message_post(
                body='dummy message 3.',
                message_type='comment',
                subtype='mail.mt_comment'
            )

            # Get the encoded message address
            encoded_msg_id = encode_msg_id(thread_msg.id, self.env)

            self.assertTrue(
                send_email.called,
                "IrMailServer.send_email wasn't called when sending outgoing email"
            )

            message = send_email.call_args[0][0]

            reply_to_address = '{}{}@{}'.format(
                MESSAGE_PREFIX,
                encoded_msg_id,
                self.outgoing_server.force_email_reply_to_domain
            )

            # Make sure the subaddress is correct in the Reply-To field
            self.assertIn(reply_to_address,
                          message['Reply-To'],
                          "Reply-To address didn't contain the correct subaddress")

            # Make sure the author name is in the Reply-To field
            self.assertIn(thread_msg.author_id.name,
                          message['Reply-To'],
                          "Reply-To address didn't contain the author name")

            self.assertIn(self.outgoing_server.force_email_from,
                          message['From'],
                          "From address didn't contain the configure From-address")