# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.

import base64
import re
from ast import literal_eval
from datetime import datetime
from unittest.mock import patch

from freezegun import freeze_time
from psycopg2 import IntegrityError
from unittest.mock import patch

from odoo.addons.base.tests.test_ir_cron import CronMixinCase
from odoo.addons.mass_mailing.tests.common import MassMailCommon
from odoo.exceptions import ValidationError
from odoo.sql_db import Cursor
from odoo.tests import Form, HttpCase, users, tagged
from odoo.tools import mute_logger

BASE_64_STRING = 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII='


@tagged("mass_mailing")
@tagged('at_install', '-post_install')  # LEGACY at_install
class TestMassMailValues(MassMailCommon):

    @classmethod
    def setUpClass(cls):
        super(TestMassMailValues, cls).setUpClass()
        cls._create_mailing_list()

    @users('user_marketing')
    def test_mailing_body_cropped_vml_image(self):
        """ Testing mail mailing responsive bg-image cropping for Outlook.

        Outlook needs background images to be converted to VML but there is no
        way to emulate `background-size: cover` that works for Windows Mail as
        well. We therefore need to crop the image in the VML version to mimick
        the style of other email clients.
        """
        attachment = {}
        def patched_get_image(self, url, session):
            return base64.b64decode(BASE_64_STRING)
        original_images_to_urls = self.env['mailing.mailing']._create_attachments_from_inline_images
        def patched_images_to_urls(self, b64images):
            urls = original_images_to_urls(b64images)
            if len(urls) == 1:
                (attachment_id, attachment_token) = re.search(r'/web/image/(?P<id>[0-9]+)\?access_token=(?P<token>.*)', urls[0]).groups()
                attachment['id'] = attachment_id
                attachment['token'] = attachment_token
                return urls
            else:
                return []
        with patch("odoo.addons.mass_mailing.models.mailing.MailingMailing._get_image_by_url",
                   new=patched_get_image), \
             patch("odoo.addons.mass_mailing.models.mailing.MailingMailing._create_attachments_from_inline_images",
                   new=patched_images_to_urls):
            mailing = self.env['mailing.mailing'].create({
                'subject': 'Test',
                'state': 'draft',
                'mailing_model_id': self.env['ir.model']._get('res.partner').id,
                'body_html': """
                    <section>
                        <!--[if mso]>
                            <v:image src="https://www.example.com/image" style="width:100px;height:100px;"/>
                        <![endif]-->
                    </section>
                """,
            })
        self.assertEqual(str(mailing.body_html).strip(), f"""
                    <section>
                        <!--[if mso]>
                            <v:image src="/web/image/{attachment['id']}?access_token={attachment['token']}" style="width:100px;height:100px;"/>
                        <![endif]-->
                    </section>
        """.strip())

    @users('user_marketing')
    def test_mailing_body_inline_image(self):
        """ Testing mail mailing base64 image conversion to attachment.

        This test ensures that the base64 images are correctly converted to
        attachments, even when they appear in MSO conditional comments.
        """
        attachments = []
        original_images_to_urls = self.env['mailing.mailing']._create_attachments_from_inline_images
        def patched_images_to_urls(self, b64images):
            urls = original_images_to_urls(b64images)
            for url in urls:
                (attachment_id, attachment_token) = re.search(r'/web/image/(?P<id>[0-9]+)\?access_token=(?P<token>.*)', url).groups()
                attachments.append({
                    'id': attachment_id,
                    'token': attachment_token,
                })
            return urls
        with patch("odoo.addons.mass_mailing.models.mailing.MailingMailing._create_attachments_from_inline_images",
                   new=patched_images_to_urls):
            mailing = self.env['mailing.mailing'].create({
                    'subject': 'Test',
                    'state': 'draft',
                    'mailing_model_id': self.env['ir.model']._get('res.partner').id,
                    'body_html': f"""
                        <section>
                            <img src="data:image/png;base64,{BASE_64_STRING}0">
                            <img src="data:image/jpg;base64,{BASE_64_STRING}1">
                            <div style='color: red; background-image:url("data:image/jpg;base64,{BASE_64_STRING}2"); display: block;'/>
                            <div style="color: red; background-image:url('data:image/jpg;base64,{BASE_64_STRING}3'); display: block;"/>
                            <div style="color: red; background-image:url(&quot;data:image/jpg;base64,{BASE_64_STRING}4&quot;); display: block;"/>
                            <div style="color: red; background-image:url(&#34;data:image/jpg;base64,{BASE_64_STRING}5&#34;); display: block;"/>
                            <div style="color: red; background-image:url(data:image/jpg;base64,{BASE_64_STRING}6); display: block;"/>
                            <div style="color: red; background-image: url(data:image/jpg;base64,{BASE_64_STRING}7); background: url('data:image/jpg;base64,{BASE_64_STRING}8'); display: block;"/>
                            <!--[if mso]>
                                <img src="data:image/png;base64,{BASE_64_STRING}9">Fake url, in text: img src="data:image/png;base64,{BASE_64_STRING}"
                                Fake url, in text: img src="data:image/png;base64,{BASE_64_STRING}"
                                <img src="data:image/jpg;base64,{BASE_64_STRING}A">
                                <div style='color: red; background-image:url("data:image/jpg;base64,{BASE_64_STRING}B"); display: block;'>Fake url, in text: style="background-image:url('data:image/png;base64,{BASE_64_STRING}');"
                                Fake url, in text: style="background-image:url('data:image/png;base64,{BASE_64_STRING}');"</div>
                                <div style="color: red; background-image:url('data:image/jpg;base64,{BASE_64_STRING}C'); display: block;"/>
                                <div style="color: red; background-image:url(&quot;data:image/jpg;base64,{BASE_64_STRING}D&quot;); display: block;"/>
                                <div style="color: red; background-image:url(&#34;data:image/jpg;base64,{BASE_64_STRING}E&#34;); display: block;"/>
                                <div style="color: red; background-image:url(data:image/jpg;base64,{BASE_64_STRING}F); display: block;"/>
                                <div style="color: red; background-image: url(data:image/jpg;base64,{BASE_64_STRING}G); background: url('data:image/jpg;base64,{BASE_64_STRING}H'); display: block;"/>
                            <![endif]-->
                            <img src="data:image/png;base64,{BASE_64_STRING}0">
                        </section>
                    """,
                })
        self.assertEqual(len(attachments), 19)
        self.assertEqual(attachments[0]['id'], attachments[18]['id'])

        # Ensure right number of routing image links exist in the final output
        found_urls = re.findall(r'/web/image/\d+\?access_token=[a-zA-Z0-9\-_=]+', mailing.body_html)
        self.assertEqual(len(found_urls), 19, "not the correct number of routing URLs.")

        # Ensure every single generated attachment URL is present somewhere
        for att in attachments:
            expected_route = f"/web/image/{att['id']}?access_token={att['token']}"
            self.assertIn(expected_route, mailing.body_html, f"Generated URL {expected_route} was dropped or mangled.")

        # False-Positive Validation:
        # The original active attributes are gone, but the "Fake url" plain text blocks must remain.
        raw_b64_count = mailing.body_html.count(BASE_64_STRING)
        self.assertEqual(raw_b64_count, 4, f"Expected 4 raw plain text instances of base64 data, found {raw_b64_count}.")

        # Verify that the MSO comment wrapper itself wasn't completely scrubbed out
        self.assertIn("Fake url, in text:", mailing.body_html)

    @users('user_marketing')
    def test_mailing_body_responsive(self):
        """ Testing mail mailing responsive mail body

        Reference: https://litmus.com/community/learning/24-how-to-code-a-responsive-email-from-scratch
        https://www.campaignmonitor.com/css/link-element/link-in-head/

        This template is meant to put inline CSS into an email's head
        """
        recipient = self.env['res.partner'].create({
            'name': 'Mass Mail Partner',
            'email': 'Customer <test.customer@example.com>',
        })
        mailing = self.env['mailing.mailing'].create({
            'subject': 'Test',
            'state': 'draft',
            'mailing_model_id': self.env['ir.model']._get('res.partner').id,
        })

        composer = self.env['mail.compose.message'].with_user(self.user_marketing).with_context({
            'default_composition_mode': 'mass_mail',
            'default_model': 'res.partner',
            'default_res_ids': recipient.ids,
        }).create({
            'subject': 'Mass Mail Responsive',
            'body': 'I am Responsive body',
            'mass_mailing_id': mailing.id
        })

        mail_values = composer._prepare_mail_values([recipient.id])
        body_html = mail_values[recipient.id]['body_html']

        self.assertIn('<!DOCTYPE html>', body_html)
        self.assertIn('<head>', body_html)
        self.assertIn('viewport', body_html)
        # This is important: we need inline css, and not <link/>
        self.assertIn('@media', body_html)
        self.assertIn('I am Responsive body', body_html)

    @users('user_marketing')
    def test_mailing_computed_fields(self):
        # Create on res.partner, with default values for computed fields
        mailing = self.env['mailing.mailing'].create({
            'subject': 'Test',
            'mailing_type': 'mail',
            'body_html': '<p>Hello <t t-out="object.name"/></p>',
            'mailing_model_id': self.env['ir.model']._get('res.partner').id,
        })
        self.assertEqual(mailing.user_id, self.user_marketing)
        self.assertEqual(mailing.medium_id, self.env.ref('utm.utm_medium_email'))
        self.assertEqual(mailing.mailing_model_name, 'res.partner')
        self.assertEqual(mailing.mailing_model_real, 'res.partner')
        self.assertEqual(mailing.reply_to_mode, 'new')
        self.assertEqual(mailing.reply_to, self.user_marketing.email_formatted)
        self.assertEqual(literal_eval(mailing.mailing_domain), [])
        # update domain
        mailing.write({
            'mailing_domain': [('email', 'ilike', 'test.example.com')]
        })
        self.assertEqual(literal_eval(mailing.mailing_domain), [('email', 'ilike', 'test.example.com')])

        # reset mailing model -> reset domain; set reply_to -> keep it
        mailing.write({
            'mailing_model_id': self.env['ir.model']._get('mailing.list').id,
            'reply_to': self.email_reply_to,
        })
        self.assertEqual(mailing.mailing_model_name, 'mailing.list')
        self.assertEqual(mailing.mailing_model_real, 'mailing.contact')
        self.assertEqual(mailing.reply_to_mode, 'new')
        self.assertEqual(mailing.reply_to, self.email_reply_to)
        # default for mailing list: depends upon contact_list_ids
        self.assertEqual(literal_eval(mailing.mailing_domain), [('list_ids', 'in', [])])
        mailing.write({
            'contact_list_ids': [(4, self.mailing_list_1.id), (4, self.mailing_list_2.id)]
        })
        self.assertEqual(literal_eval(mailing.mailing_domain), [('list_ids', 'in', (self.mailing_list_1 | self.mailing_list_2).ids)])

        # reset mailing model -> reset domain and reply to mode
        mailing.write({
            'mailing_model_id': self.env['ir.model']._get('discuss.channel').id,
        })
        self.assertEqual(mailing.mailing_model_name, 'discuss.channel')
        self.assertEqual(mailing.mailing_model_real, 'discuss.channel')
        self.assertEqual(mailing.reply_to_mode, 'update')
        self.assertFalse(mailing.reply_to)

    @users('user_marketing')
    def test_mailing_computed_fields_domain_w_filter(self):
        """ Test domain update, involving mailing.filters added in 15.1. """
        # Create on res.partner, with default values for computed fields
        mailing = self.env['mailing.mailing'].create({
            'subject': 'Test',
            'mailing_type': 'mail',
            'body_html': '<p>Hello <t t-out="object.name"/></p>',
            'mailing_model_id': self.env['ir.model']._get('res.partner').id,
        })
        self.assertEqual(literal_eval(mailing.mailing_domain), [])

        # prepare initial data
        filter_1, filter_2, filter_3 = self.env['mailing.filter'].create([
            {'name': 'General channel',
             'mailing_domain' : [('name', '=', 'general')],
             'mailing_model_id': self.env['ir.model']._get('discuss.channel').id,
            },
            {'name': 'LLN City',
             'mailing_domain' : [('city', 'ilike', 'LLN')],
             'mailing_model_id': self.env['ir.model']._get('res.partner').id,
            },
            {'name': 'Email based',
             'mailing_domain' : [('email', 'ilike', 'info@odoo.com')],
             'mailing_model_id': self.env['ir.model']._get('res.partner').id,
            }
        ])

        # check that adding mailing_filter_id updates domain correctly
        mailing.mailing_filter_id = filter_2
        self.assertEqual(literal_eval(mailing.mailing_domain), literal_eval(filter_2.mailing_domain))

        # cannot set a filter linked to another model
        with self.assertRaises(ValidationError):
            mailing.mailing_filter_id = filter_1

        # resetting model should reset domain, even if filter was chosen previously
        mailing.mailing_model_id = self.env['ir.model']._get('discuss.channel').id
        self.assertEqual(literal_eval(mailing.mailing_domain), [])

        # changing the filter should update the mailing domain correctly
        mailing.mailing_filter_id = filter_1
        self.assertEqual(literal_eval(mailing.mailing_domain), literal_eval(filter_1.mailing_domain))

        # changing the domain should not empty the mailing_filter_id
        mailing.mailing_domain = "[('email', 'ilike', 'info_be@odoo.com')]"
        self.assertEqual(mailing.mailing_filter_id, filter_1, "Filter should not be unset even if domain is changed")

        # deleting the filter record should not delete the domain on mailing
        mailing.mailing_model_id = self.env['ir.model']._get('res.partner').id
        mailing.mailing_filter_id = filter_3
        filter_3_domain = filter_3.mailing_domain
        self.assertEqual(literal_eval(mailing.mailing_domain), literal_eval(filter_3_domain))
        filter_3.unlink()  # delete the filter record
        self.assertFalse(mailing.mailing_filter_id, "Should unset filter if it is deleted")
        self.assertEqual(literal_eval(mailing.mailing_domain), literal_eval(filter_3_domain), "Should still have the same domain")

    @users('user_marketing')
    def test_mailing_computed_fields_default(self):
        mailing = self.env['mailing.mailing'].with_context(
            default_mailing_domain=repr([('email', 'ilike', 'test.example.com')])
        ).create({
            'subject': 'Test',
            'mailing_type': 'mail',
            'body_html': '<p>Hello <t t-out="object.name"/></p>',
            'mailing_model_id': self.env['ir.model']._get('res.partner').id,
        })
        self.assertEqual(literal_eval(mailing.mailing_domain), [('email', 'ilike', 'test.example.com')])

    @users('user_marketing')
    def test_mailing_computed_fields_default_email_from(self):
        # Testing if the email_from is correctly computed when an
        # alias domain for the company is set

        # Setup mail outgoing server for use cases

        from_filter_match, from_filter_missmatch = self.env['ir.mail_server'].sudo().create([
            # Case where alias domain is set and there is a default outgoing email server
            # for mass mailing. from_filter matches domain of company alias domain
            # before record creation
            {
                    'name' : 'mass_mailing_test_match_from_filter',
                    'from_filter' : self.alias_domain,
                    'smtp_host' : 'not_real@smtp.com',
            },
            # Case where alias domain is set and there is a default outgoing email server
            # for mass mailing. from_filter DOES NOT match domain of company alias domain
            # before record creation
            {
                    'name' : 'mass_mailing_test_from_missmatch',
                    'from_filter' : 'test.com',
                    'smtp_host' : 'not_real@smtp.com',
            },
        ])

        # Expected combos of server vs FROM values

        servers = [
            self.env['ir.mail_server'],
            from_filter_match,
            from_filter_missmatch,
        ]
        expected_from_all = [
            self.env.user.email_formatted,  # default when no server
            self.env.user.company_id.alias_domain_id.default_from_email,  # matches company alias domain
            self.env.user.email_formatted,  # not matching from filter -> back to user from
        ]

        for mail_server, expected_from in zip(servers, expected_from_all):
            with self.subTest(server_name=mail_server.name):
                # When a mail server is set, we update the mass mailing
                # settings to designate a dedicated outgoing email server
                if mail_server:
                    self.env['res.config.settings'].sudo().create({
                        'mass_mailing_mail_server_id' : mail_server.id,
                        'mass_mailing_outgoing_mail_server' : mail_server,
                    }).execute()

                # Create mailing
                mailing = self.env['mailing.mailing'].create({
                    'subject': f'Test {mail_server.name}',
                })

                # Check email_from
                self.assertEqual(mailing.email_from, expected_from)

                # If configured, check if dedicated email outgoing server is
                # on mailing record
                self.assertEqual(mailing.mail_server_id, mail_server)

    @users('user_marketing')
    def test_mailing_computed_fields_form(self):
        mailing_form = Form(self.env['mailing.mailing'].with_context(
            default_mailing_domain="[('email', 'ilike', 'test.example.com')]",
            default_mailing_model_id=self.env['ir.model']._get('res.partner').id,
        ))
        self.assertEqual(
            literal_eval(mailing_form.mailing_domain),
            [('email', 'ilike', 'test.example.com')],
        )
        self.assertEqual(mailing_form.mailing_model_real, 'res.partner')

    @users('user_marketing')
    def test_mailing_create_on_send(self):
        recipient = self.env['res.partner'].create({
            'name': 'Mass Mail Partner',
            'email': 'Customer <test.customer@example.com>',
        })

        composer = self.env['mail.compose.message'].with_user(self.user_marketing).with_context({
            'default_composition_mode': 'mass_mail',
            'default_model': 'res.partner',
            'default_res_ids': recipient.ids,
        }).create({
            'subject': 'Mass Mail Responsive',
            'body': 'I am Responsive body',
            'mass_mailing_create': True,
        })
        self.assertFalse(composer.mass_mailing_id, "No mailing should've been created")

        with self.mock_mail_gateway():
            composer._action_send_mail(recipient.ids)

        self.assertEqual(len(composer.mass_mailing_id.ids), 1, "A mailing should've been created")
        self.assertEqual(
            composer.mass_mailing_id.subject,
            "Mass Mail Responsive",
        )

        mail_values = composer._prepare_mail_values(recipient.ids)[recipient.id]
        self.assertIn(
            "Received the mailing <b>Mass Mail Responsive</b>",
            mail_values["body"],
            "The composer doesn't use the provided subject"
        )

    @mute_logger('odoo.sql_db')
    @users('user_marketing')
    def test_mailing_trace_values(self):
        recipient = self.partner_employee

        # both void and 0 are invalid, document should have an id != 0
        with self.assertRaises(IntegrityError):
            self.env['mailing.trace'].create({
                'model': recipient._name,
            })
        with self.assertRaises(IntegrityError):
            self.env['mailing.trace'].create({
                'model': recipient._name,
                'res_id': 0,
            })
        with self.assertRaises(IntegrityError):
            self.env['mailing.trace'].create({
                'res_id': 3,
            })

        activity = self.env['mailing.trace'].create({
            'model': recipient._name,
            'res_id': recipient.id,
        })
        with self.assertRaises(IntegrityError):
            activity.write({'model': False})
            self.env.flush_all()
        with self.assertRaises(IntegrityError):
            activity.write({'res_id': False})
            self.env.flush_all()
        with self.assertRaises(IntegrityError):
            activity.write({'res_id': 0})
            self.env.flush_all()

    def test_mailing_editor_created_attachments(self):
        mailing = self.env['mailing.mailing'].create({
            'subject': 'Test',
            'mailing_type': 'mail',
            'body_html': '<p>Hello</p>',
            'mailing_model_id': self.env['ir.model']._get('res.partner').id,
        })
        blob = b'blob1'

        # Created when uploading an image
        original_svg_attachment = self.env['ir.attachment'].create({
            "name": "test SVG",
            "mimetype": "image/svg+xml",
            "raw": blob,
            "public": True,
            "res_model": "mailing.mailing",
            "res_id": mailing.id,
        })

        # Created when saving the mass_mailing
        png_duplicate_of_svg_attachment = self.env['ir.attachment'].create({
            "name": "test SVG duplicate",
            "mimetype": "image/png",
            "raw": blob,
            "public": True,
            "res_model": "mailing.mailing",
            "res_id": mailing.id,
            "original_id": original_svg_attachment.id
        })

        # Created by uploading new image
        original_png_attachment = self.env['ir.attachment'].create({
            "name": "test PNG",
            "mimetype": "image/png",
            "raw": blob,
            "public": True,
            "res_model": "mailing.mailing",
            "res_id": mailing.id,
        })

        # Created by modify_image in editor controller
        self.env['ir.attachment'].create({
            "name": "test PNG duplicate",
            "mimetype": "image/png",
            "raw": blob,
            "public": True,
            "res_model": "mailing.mailing",
            "res_id": mailing.id,
            "original_id": original_png_attachment.id
        })

        mail_thread_attachments = mailing._get_mail_thread_data_attachments()
        self.assertSetEqual(set(mail_thread_attachments.ids), {png_duplicate_of_svg_attachment.id, original_png_attachment.id})

    @users('user_marketing')
    @mute_logger('odoo.addons.mail.models.mail_mail')
    def test_process_mailing_queue_without_html_body(self):
        """ Test mailing with past schedule date and without any html body """
        mailing = self.env['mailing.mailing'].create({
                'subject': 'some subject',
                'mailing_model_id': self.env['ir.model']._get('res.partner').id,
                'preview': "Check it out before its too late",
                'body_html': False,
                'schedule_date': datetime(2023, 2, 17, 11, 0),
            })
        mailing.action_put_in_queue()
        with self.mock_mail_gateway(mail_unlink_sent=False), self.enter_registry_test_mode():
            self.env.ref('mass_mailing.ir_cron_mass_mailing_queue').sudo().method_direct_trigger()

        self.assertFalse(mailing.body_html)
        self.assertEqual(mailing.mailing_model_name, 'res.partner')


@tagged("mass_mailing")
class TestMassMailingMailServer(MassMailCommon):

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.public_server = cls.env['ir.mail_server'].sudo().create({
            'from_filter': 'test.mycompany.com',
            'name': 'public_server',
            'smtp_host': 'public.example.com',
            'smtp_user': 'public@test.mycompany.com',
        })

    def test_owner_allowed_when_server_used_by_done_mailing(self):
        """A server used only by a finished mailing can still be turned into
        a personal server."""
        mailing = self.env['mailing.mailing'].create({
            'body_html': '<p>x</p>',
            'mail_server_id': self.public_server.id,
            'mailing_model_id': self.env['ir.model']._get_id('res.partner'),
            'subject': 'S',
        })
        mailing.write({'state': 'done'})

        self.public_server.owner_user_id = self.user_marketing
        self.assertEqual(self.public_server.owner_user_id, self.user_marketing)

    def test_owner_blocked_when_server_is_dedicated_default(self):
        """A server set as the Email Marketing dedicated server cannot be
        turned into a personal server."""
        self.env['ir.config_parameter'].sudo().set_int(
            'mass_mailing.mail_server_id', self.public_server.id)

        with self.assertRaises(ValidationError):
            self.public_server.owner_user_id = self.user_marketing

    def test_owner_blocked_when_server_used_by_mailing(self):
        """A server already in use by a queued mailing cannot be turned into
        a personal server."""
        mailing = self.env['mailing.mailing'].create({
            'body_html': '<p>x</p>',
            'mail_server_id': self.public_server.id,
            'mailing_model_id': self.env['ir.model']._get_id('res.partner'),
            'subject': 'S',
        })
        mailing.action_put_in_queue()

        with self.assertRaises(ValidationError):
            self.public_server.owner_user_id = self.user_marketing

    def test_fallback_excludes_personal_mail_server(self):
        """When no dedicated server is configured, the personal server must
        not be picked even if its from_filter matches the campaign sender."""
        # Wipe pre-existing servers so _find_mail_server only sees the two we set up.
        self.env['ir.mail_server'].sudo().search([('id', '!=', self.public_server.id)]).unlink()
        self.env['ir.mail_server'].sudo().create({
            'from_filter': 'user_marketing@test.mycompany.com',
            'name': 'personal_server',
            'owner_user_id': self.user_marketing.id,
            'smtp_host': 'personal.example.com',
            'smtp_user': 'user_marketing@test.mycompany.com',
        })

        recipient = self.env['res.partner'].create({
            'email': 'target@test.mycompany.com',
            'name': 'Target',
        })
        mailing = self.env['mailing.mailing'].create({
            'body_html': '<p>x</p>',
            'email_from': 'user_marketing@test.mycompany.com',
            'mailing_domain': [('id', '=', recipient.id)],
            'mailing_model_id': self.env['ir.model']._get_id('res.partner'),
            'subject': 'S',
            'user_id': self.user_marketing.id,
        })

        with self.mock_smtplib_connection():
            mailing.with_user(self.user_marketing).action_send_mail()

        self.connect_mocked.assert_called_once()
        self.assertEqual(
            self.connect_mocked.call_args.kwargs['mail_server_id'],
            self.public_server.id,
        )


@tagged('mass_mailing')
@tagged('at_install', '-post_install')  # LEGACY at_install
class TestMassMailFeatures(MassMailCommon, CronMixinCase):

    @classmethod
    def setUpClass(cls):
        super(TestMassMailFeatures, cls).setUpClass()
        cls._create_mailing_list()

    @users('user_marketing')
    @mute_logger('odoo.addons.mail.models.mail_mail')
    def test_mailing_cron_trigger(self):
        """ Technical test to ensure the cron is triggered at the correct
        time """

        cron_id = self.env.ref('mass_mailing.ir_cron_mass_mailing_queue').id
        partner = self.env['res.partner'].create({
            'name': 'Jean-Alphonce',
            'email': 'jeanalph@example.com',
        })
        common_mailing_values = {
            'subject': "Who's there?",
            'mailing_model_id': self.env['ir.model']._get('res.partner').id,
            'mailing_domain': [('id', '=', partner.id)],
            'body_html': 'The marketing mailing test.',
            'schedule_type': 'scheduled',
        }

        now = datetime(2021, 2, 5, 16, 43, 20)
        then = datetime(2021, 2, 7, 12, 0, 0)

        with freeze_time(now):
            for (test, truth) in [(False, now), (then, then)]:
                with self.subTest(schedule_date=test):
                    with self.capture_triggers(cron_id) as capt:
                        mailing = self.env['mailing.mailing'].create({
                            **common_mailing_values,
                            'schedule_date': test,
                        })
                        mailing.action_put_in_queue()
                    capt.records.ensure_one()
                    self.assertLessEqual(capt.records.call_at, truth)

    @users('user_marketing')
    @mute_logger('odoo.addons.mail.models.mail_mail')
    def test_mailing_deletion(self):
        """ Test deletion in various use case, depending on reply-to """
        # 1- Keep archives and reply-to set to 'answer = new thread'
        mailing = self.env['mailing.mailing'].create({
            'subject': 'TestDeletion',
            'body_html': "<div>Hello {object.name}</div>",
            'mailing_model_id': self.env['ir.model']._get('mailing.list').id,
            'contact_list_ids': [(6, 0, self.mailing_list_1.ids)],
            'keep_archives': True,
            'reply_to_mode': 'new',
            'reply_to': self.email_reply_to,
        })
        self.assertEqual(self.mailing_list_1.contact_ids.message_ids, self.env['mail.message'])

        with self.mock_mail_gateway(mail_unlink_sent=True):
            mailing.action_send_mail()

        self.assertEqual(len(self._mails), 3)
        self.assertEqual(len(self._new_mails.exists()), 3)
        self.assertEqual(len(self.mailing_list_1.contact_ids.message_ids), 3)

        # 2- Keep archives and reply-to set to 'answer = update thread'
        self.mailing_list_1.contact_ids.message_ids.unlink()
        mailing = mailing.copy()
        mailing.write({
            'reply_to_mode': 'update',
        })
        self.assertEqual(self.mailing_list_1.contact_ids.message_ids, self.env['mail.message'])

        with self.mock_mail_gateway(mail_unlink_sent=True):
            mailing.action_send_mail()

        self.assertEqual(len(self._mails), 3)
        self.assertEqual(len(self._new_mails.exists()), 3)
        self.assertEqual(len(self.mailing_list_1.contact_ids.message_ids), 3)

        # 3- Remove archives and reply-to set to 'answer = new thread'
        self.mailing_list_1.contact_ids.message_ids.unlink()
        mailing = mailing.copy()
        mailing.write({
            'keep_archives': False,
            'reply_to_mode': 'new',
            'reply_to': self.email_reply_to,
        })
        self.assertEqual(self.mailing_list_1.contact_ids.message_ids, self.env['mail.message'])

        with self.mock_mail_gateway(mail_unlink_sent=True):
            mailing.action_send_mail()

        self.assertEqual(len(self._mails), 3)
        self.assertEqual(len(self._new_mails.exists()), 0)
        self.assertEqual(self.mailing_list_1.contact_ids.message_ids, self.env['mail.message'])

        # 4- Remove archives and reply-to set to 'answer = update thread'
        # Imply keeping mail.message for gateway reply)
        self.mailing_list_1.contact_ids.message_ids.unlink()
        mailing = mailing.copy()
        mailing.write({
            'keep_archives': False,
            'reply_to_mode': 'update',
        })
        self.assertEqual(self.mailing_list_1.contact_ids.message_ids, self.env['mail.message'])

        with self.mock_mail_gateway(mail_unlink_sent=True):
            mailing.action_send_mail()

        self.assertEqual(len(self._mails), 3)
        self.assertEqual(len(self._new_mails.exists()), 0)
        self.assertEqual(len(self.mailing_list_1.contact_ids.message_ids), 3)

    @users('user_marketing')
    @mute_logger('odoo.addons.mail.models.mail_mail')
    def test_mailing_on_res_partner(self):
        """ Test mailing on res.partner model: ensure default recipients are
        correctly computed """
        partner_a = self.env['res.partner'].create({
            'name': 'test email 1',
            'email': 'test1@example.com',
        })
        partner_b = self.env['res.partner'].create({
            'name': 'test email 2',
            'email': 'test2@example.com',
        })
        self.env['mail.blacklist'].create({'email': 'Test2@example.com',})

        mailing = self.env['mailing.mailing'].create({
            'subject': 'One',
            'mailing_model_id': self.env['ir.model']._get('res.partner').id,
            'mailing_domain': [('id', 'in', (partner_a | partner_b).ids)],
            'body_html': 'This is mass mail marketing demo'
        })
        self.assertEqual(mailing.user_id, self.user_marketing)
        mailing.action_put_in_queue()
        self.assertEqual(mailing.email_from, self.env.user.email_formatted)
        with self.mock_mail_gateway(mail_unlink_sent=False), self.enter_registry_test_mode():
            self.env.ref('mass_mailing.ir_cron_mass_mailing_queue').sudo().method_direct_trigger()

        author = self.user_marketing.partner_id
        email_values = {'email_from': mailing.email_from}
        self.assertMailTraces(
            [{'partner': partner_a, 'email_values': email_values},
             {'partner': partner_b, 'trace_status': 'cancel', 'failure_type': 'mail_bl', 'email_values': email_values}],
            mailing, partner_a + partner_b,
            check_mail=True, author=author,
        )

    @users('user_marketing')
    @mute_logger('odoo.addons.mail.models.mail_mail')
    def test_mailing_shortener(self):
        mailing = self.env['mailing.mailing'].create({
            'subject': 'TestShortener',
            'body_html': """<div>
Hi,
<t t-set="url" t-value="'www.odoo.com'"/>
<t t-set="httpurl" t-value="'https://www.odoo.eu'"/>
Website0: <a id="url0" t-attf-href="https://www.odoo.tz/my/{{object.name}}">https://www.odoo.tz/my/<t t-out="object.name"/></a>
Website1: <a id="url1" href="https://www.odoo.be">https://www.odoo.be</a>
Website2: <a id="url2" t-attf-href="https://{{url}}">https://<t t-out="url"/></a>
Website3: <a id="url3" t-att-href="httpurl"><t t-out="httpurl"/></a>
External1: <a id="url4" href="https://www.example.com/foo/bar?baz=qux">Youpie</a>
Email: <a id="url5" href="mailto:test@odoo.com">test@odoo.com</a></div>""",
            'mailing_model_id': self.env['ir.model']._get('mailing.list').id,
            'reply_to_mode': 'new',
            'reply_to': self.email_reply_to,
            'contact_list_ids': [(6, 0, self.mailing_list_1.ids)],
            'keep_archives': True,
        })

        mailing.action_put_in_queue()

        with self.mock_mail_gateway(mail_unlink_sent=False), self.enter_registry_test_mode():
            self.env.ref('mass_mailing.ir_cron_mass_mailing_queue').sudo().method_direct_trigger()

        author = self.user_marketing.partner_id
        email_values = {'email_from': mailing.email_from}
        self.assertMailTraces(
            [{'email': 'fleurus@example.com', 'email_values': email_values},
             {'email': 'gorramts@example.com', 'email_values': email_values},
             {'email': 'ybrant@example.com', 'email_values': email_values}],
            mailing, self.mailing_list_1.contact_ids,
            check_mail=True, author=author,
        )

        for contact in self.mailing_list_1.contact_ids:
            new_mail = self._find_mail_mail_wrecord(contact)
            for link_info in [('url0', 'https://www.odoo.tz/my/%s' % contact.name, True),
                              ('url1', 'https://www.odoo.be', True),
                              ('url2', 'https://www.odoo.com', True),
                              ('url3', 'https://www.odoo.eu', True),
                              ('url4', 'https://www.example.com/foo/bar?baz=qux', True),
                              ('url5', 'mailto:test@odoo.com', False)]:
                # TDE FIXME: why going to mail message id ? mail.body_html seems to fail, check
                link_params = {
                    'utm_medium': 'Email',
                    'utm_source': 'Mass Mailing',
                    'utm_reference': f'mailing.mailing,{mailing.id}',
                }
                if link_info[0] == 'url4':
                    link_params['baz'] = 'qux'
                self.assertLinkShortenedHtml(
                    new_mail.mail_message_id.body,
                    link_info,
                    link_params=link_params,
                )


@tagged("mail_mail")
@tagged('at_install', '-post_install')  # LEGACY at_install
class TestMailingHeaders(MassMailCommon, HttpCase):
    """ Test headers + linked controllers """

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls._create_mailing_list()
        cls.test_mailing = cls.env['mailing.mailing'].with_user(cls.user_marketing).create({
            "body_html": """
<p>Hello <t t-out="object.name"/>
    <a href="/unsubscribe_from_list">UNSUBSCRIBE</a>
    <a href="/view">VIEW</a>
</p>""",
            "contact_list_ids": [(4, cls.mailing_list_1.id)],
            "mailing_model_id": cls.env["ir.model"]._get("mailing.list").id,
            "mailing_type": "mail",
            "subject": "Test for {{ object.name }}",
        })

    @users('user_marketing')
    def test_mailing_unsubscribe_headers(self):
        """ Check unsubscribe headers are present in outgoing emails and work
        as one-click """
        test_mailing = self.test_mailing.with_env(self.env)
        test_mailing.action_put_in_queue()

        with self.mock_mail_gateway(mail_unlink_sent=False):
            test_mailing.action_send_mail()

        for contact in self.mailing_list_1.contact_ids:
            new_mail = self._find_mail_mail_wrecord(contact)
            # check mail.mail still have default links
            self.assertIn("/unsubscribe_from_list", new_mail.body)
            self.assertIn("/view", new_mail.body)

            # check outgoing email headers (those are put into outgoing email
            # not in the mail.mail record)
            email = self._find_sent_email_wemail(contact.email)
            headers = email.get("headers")
            unsubscribe_oneclick_url = test_mailing._get_unsubscribe_oneclick_url(contact.email, contact.id)
            self.assertTrue(headers, "Mass mailing emails should have headers for unsubscribe")
            self.assertEqual(headers.get("List-Unsubscribe"), f"<{unsubscribe_oneclick_url}>")
            self.assertEqual(headers.get("List-Unsubscribe-Post"), "List-Unsubscribe=One-Click")
            self.assertEqual(headers.get("Precedence"), "list")

            # check outgoing email has real links
            self.assertNotIn("/unsubscribe_from_list", email["body"])

            # unsubscribe in one-click
            unsubscribe_oneclick_url = headers["List-Unsubscribe"].strip("<>")
            self.url_open(unsubscribe_oneclick_url, method='POST')

            # should be unsubscribed
            self.assertTrue(contact.subscription_ids.opt_out)


@tagged('at_install', '-post_install')  # LEGACY at_install
class TestMailingScheduleDateWizard(MassMailCommon):

    @mute_logger('odoo.addons.mail.models.mail_mail')
    @users('user_marketing')
    def test_mailing_schedule_date(self):
        mailing = self.env['mailing.mailing'].create({
            'subject': 'some subject'
        })
        # create a schedule date wizard
        wizard_form = Form(
            self.env['mailing.mailing.schedule.date'].with_context(default_mass_mailing_id=mailing.id))

        # set a schedule date
        wizard_form.schedule_date = datetime(2021, 4, 30, 9, 0)
        wizard = wizard_form.save()
        wizard.action_schedule_date()

        # assert that the schedule_date and schedule_type fields are correct and that the mailing is put in queue
        self.assertEqual(mailing.schedule_date, datetime(2021, 4, 30, 9, 0))
        self.assertEqual(mailing.schedule_type, 'scheduled')
        self.assertEqual(mailing.state, 'in_queue')


@tagged('at_install', '-post_install')  # LEGACY at_install
class TestMassMailingActions(MassMailCommon):
    def test_mailing_action_open(self):
        mass_mailings = self.env['mailing.mailing'].create([
            {'subject': 'First subject'},
            {'subject': 'Second subject'}
        ])
        # Create two traces: one linked to the created mass.mailing and one not (action should open only the first)
        self.env["mailing.trace"].create([{
                "trace_status": "open",
                "mass_mailing_id": mass_mailings[0].id,
                "model": "res.partner",
                "res_id": self.partner_admin.id,
            }, {
                "trace_status": "open",
                "mass_mailing_id": mass_mailings[1].id,
                "model": "res.partner",
                "res_id": self.partner_employee.id,
            }
        ])
        results = mass_mailings[0].action_view_opened()
        results_partner = self.env["res.partner"].search(results['domain'])
        self.assertEqual(results_partner, self.partner_admin, "Trace leaked from mass_mailing_2 to mass_mailing_1")
