Thibault Delavallée
c98f259736
In this commit we add a method to check MailMail, based on a given record. When having duplicates to differentiate in a given mailing, having just recipients is not sufficient as multiple emails may match a given recipients list. Checking model / res_id is another method for finding emails. This will be used notably to add tests for exclusion list and duplicates management in standard mail composer, outside of mass mailing context. While being at it, use subTests when having loops checking values. That way it is easier to fix tests that have several failing values in the same global assertMailMail_* . Task-3132710 (Mail: Configurable composer) Part-of: odoo/odoo#99482
281 lines
13 KiB
Python
281 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
|
|
|
import datetime
|
|
import random
|
|
import re
|
|
import werkzeug
|
|
|
|
from unittest.mock import patch
|
|
|
|
from odoo import tools
|
|
from odoo.addons.link_tracker.tests.common import MockLinkTracker
|
|
from odoo.addons.mail.tests.common import MailCase, MailCommon, mail_new_test_user
|
|
from odoo.sql_db import Cursor
|
|
|
|
|
|
class MassMailCase(MailCase, MockLinkTracker):
|
|
|
|
# ------------------------------------------------------------
|
|
# ASSERTS
|
|
# ------------------------------------------------------------
|
|
|
|
def assertMailingStatistics(self, mailing, **kwargs):
|
|
""" Helper to assert mailing statistics fields. As we have many of them
|
|
it helps lessening test asserts. """
|
|
if not kwargs.get('expected'):
|
|
kwargs['expected'] = len(mailing.mailing_trace_ids)
|
|
if not kwargs.get('delivered'):
|
|
kwargs['delivered'] = len(mailing.mailing_trace_ids)
|
|
for fname in ['scheduled', 'expected', 'sent', 'delivered',
|
|
'opened', 'replied', 'clicked',
|
|
'canceled', 'failed', 'bounced']:
|
|
self.assertEqual(
|
|
mailing[fname], kwargs.get(fname, 0),
|
|
'Mailing %s statistics failed: got %s instead of %s' % (fname, mailing[fname], kwargs.get(fname, 0))
|
|
)
|
|
|
|
def assertMailTraces(self, recipients_info, mailing, records,
|
|
check_mail=True, sent_unlink=False, author=None,
|
|
mail_links_info=None):
|
|
""" Check content of traces. Traces are fetched based on a given mailing
|
|
and records. Their content is compared to recipients_info structure that
|
|
holds expected information. Links content may be checked, notably to
|
|
assert shortening or unsubscribe links. Mail.mail records may optionally
|
|
be checked.
|
|
|
|
:param recipients_info: list[{
|
|
# TRACE
|
|
'partner': res.partner record (may be empty),
|
|
'email': email used when sending email (may be empty, computed based on partner),
|
|
'trace_status': outgoing / sent / open / reply / bounce / error / cancel (sent by default),
|
|
'record: linked record,
|
|
# MAIL.MAIL
|
|
'content': optional content that should be present in mail.mail body_html;
|
|
'failure_type': optional failure reason;
|
|
}, { ... }]
|
|
|
|
:param mailing: a mailing.mailing record from which traces have been
|
|
generated;
|
|
:param records: records given to mailing that generated traces. It is
|
|
used notably to find traces using their IDs;
|
|
:param check_mail: if True, also check mail.mail records that should be
|
|
linked to traces;
|
|
:param sent_unlink: it True, sent mail.mail are deleted and we check gateway
|
|
output result instead of actual mail.mail records;
|
|
:param mail_links_info: if given, should follow order of ``recipients_info``
|
|
and give details about links. See ``assertLinkShortenedHtml`` helper for
|
|
more details about content to give;
|
|
:param author: author of sent mail.mail;
|
|
"""
|
|
# map trace state to email state
|
|
state_mapping = {
|
|
'sent': 'sent',
|
|
'open': 'sent', # opened implies something has been sent
|
|
'reply': 'sent', # replied implies something has been sent
|
|
'error': 'exception',
|
|
'cancel': 'cancel',
|
|
'bounce': 'cancel',
|
|
}
|
|
|
|
traces = self.env['mailing.trace'].search([
|
|
('mass_mailing_id', 'in', mailing.ids),
|
|
('res_id', 'in', records.ids)
|
|
])
|
|
|
|
# ensure trace coherency
|
|
self.assertTrue(all(s.model == records._name for s in traces))
|
|
self.assertEqual(set(s.res_id for s in traces), set(records.ids))
|
|
|
|
# check each traces
|
|
if not mail_links_info:
|
|
mail_links_info = [None] * len(recipients_info)
|
|
for recipient_info, link_info, record in zip(recipients_info, mail_links_info, records):
|
|
partner = recipient_info.get('partner', self.env['res.partner'])
|
|
email = recipient_info.get('email')
|
|
status = recipient_info.get('trace_status', 'sent')
|
|
record = record or recipient_info.get('record')
|
|
content = recipient_info.get('content')
|
|
if email is None and partner:
|
|
email = partner.email_normalized
|
|
|
|
recipient_trace = traces.filtered(
|
|
lambda t: (t.email == email or (not email and not t.email)) and \
|
|
t.trace_status == status and \
|
|
(t.res_id == record.id if record else True)
|
|
)
|
|
self.assertTrue(
|
|
len(recipient_trace) == 1,
|
|
'MailTrace: email %s (recipient %s, status: %s, record: %s): found %s records (1 expected)' % (email, partner, status, record, len(recipient_trace))
|
|
)
|
|
self.assertTrue(bool(recipient_trace.mail_mail_id_int))
|
|
if 'failure_type' in recipient_info or status in ('error', 'cancel', 'bounce'):
|
|
self.assertEqual(recipient_trace.failure_type, recipient_info['failure_type'])
|
|
|
|
if check_mail:
|
|
if author is None:
|
|
author = self.env.user.partner_id
|
|
|
|
# mail.mail specific values to check
|
|
fields_values = {'mailing_id': mailing}
|
|
|
|
# specific for partner: email_formatted is used
|
|
if partner:
|
|
if status == 'sent' and sent_unlink:
|
|
self.assertSentEmail(author, [partner])
|
|
else:
|
|
self.assertMailMail(partner, state_mapping[status], author=author, content=content, fields_values=fields_values)
|
|
# specific if email is False -> could have troubles finding it if several falsy traces
|
|
elif not email and status in ('cancel', 'bounce'):
|
|
self.assertMailMailWId(recipient_trace.mail_mail_id_int, state_mapping[status], author=author, content=content, fields_values=fields_values)
|
|
else:
|
|
self.assertMailMailWEmails([email], state_mapping[status], author=author, content=content, fields_values=fields_values)
|
|
|
|
if link_info:
|
|
trace_mail = self._find_mail_mail_wrecord(record)
|
|
for (anchor_id, url, is_shortened, add_link_params) in link_info:
|
|
link_params = {'utm_medium': 'Email', 'utm_source': mailing.name}
|
|
if add_link_params:
|
|
link_params.update(**add_link_params)
|
|
self.assertLinkShortenedHtml(
|
|
trace_mail.body_html,
|
|
(anchor_id, url, is_shortened),
|
|
link_params=link_params,
|
|
)
|
|
|
|
# ------------------------------------------------------------
|
|
# TOOLS
|
|
# ------------------------------------------------------------
|
|
|
|
def gateway_mail_bounce(self, mailing, record, bounce_base_values=None):
|
|
""" Generate a bounce at mailgateway level.
|
|
|
|
:param mailing: a ``mailing.mailing`` record on which we find a trace
|
|
to bounce;
|
|
:param record: record which should bounce;
|
|
:param bounce_base_values: optional values given to routing;
|
|
"""
|
|
trace = mailing.mailing_trace_ids.filtered(lambda t: t.model == record._name and t.res_id == record.id)
|
|
|
|
parsed_bounce_values = {
|
|
'email_from': 'some.email@external.example.com', # TDE check: email_from -> trace email ?
|
|
'to': 'bounce@test.example.com', # TDE check: bounce alias ?
|
|
'message_id': tools.generate_tracking_message_id('MailTest'),
|
|
'bounced_partner': self.env['res.partner'].sudo(),
|
|
'bounced_message': self.env['mail.message'].sudo()
|
|
}
|
|
if bounce_base_values:
|
|
parsed_bounce_values.update(bounce_base_values)
|
|
parsed_bounce_values.update({
|
|
'bounced_email': trace.email,
|
|
'bounced_msg_id': [trace.message_id],
|
|
})
|
|
self.env['mail.thread']._routing_handle_bounce(False, parsed_bounce_values)
|
|
|
|
def gateway_mail_click(self, mailing, record, click_label):
|
|
""" Simulate a click on a sent email. """
|
|
trace = mailing.mailing_trace_ids.filtered(lambda t: t.model == record._name and t.res_id == record.id)
|
|
email = self._find_sent_mail_wemail(trace.email)
|
|
self.assertTrue(bool(email))
|
|
for (_url_href, link_url, _dummy, label) in re.findall(tools.HTML_TAG_URL_REGEX, email['body']):
|
|
if label == click_label and '/r/' in link_url: # shortened link, like 'http://localhost:8069/r/LBG/m/53'
|
|
parsed_url = werkzeug.urls.url_parse(link_url)
|
|
path_items = parsed_url.path.split('/')
|
|
code, trace_id = path_items[2], int(path_items[4])
|
|
self.assertEqual(trace.id, trace_id)
|
|
|
|
self.env['link.tracker.click'].sudo().add_click(
|
|
code,
|
|
ip='100.200.300.%3f' % random.random(),
|
|
country_code='BE',
|
|
mailing_trace_id=trace.id
|
|
)
|
|
break
|
|
else:
|
|
raise AssertionError('url %s not found in mailing %s for record %s' % (click_label, mailing, record))
|
|
|
|
@classmethod
|
|
def _create_bounce_trace(cls, mailing, records, dt=None):
|
|
if dt is None:
|
|
dt = datetime.datetime.now() - datetime.timedelta(days=1)
|
|
return cls._create_traces(mailing, records, dt, trace_status='bounce')
|
|
|
|
@classmethod
|
|
def _create_sent_traces(cls, mailing, records, dt=None):
|
|
if dt is None:
|
|
dt = datetime.datetime.now() - datetime.timedelta(days=1)
|
|
return cls._create_traces(mailing, records, dt, trace_status='sent')
|
|
|
|
@classmethod
|
|
def _create_traces(cls, mailing, records, dt, **values):
|
|
if 'email_normalized' in records:
|
|
fname = 'email_normalized'
|
|
elif 'email_from' in records:
|
|
fname = 'email_from'
|
|
else:
|
|
fname = 'email'
|
|
randomized = random.random()
|
|
# Cursor.now() uses transaction's timestamp and not datetime lib -> freeze_time
|
|
# is not sufficient
|
|
with patch.object(Cursor, 'now', lambda *args, **kwargs: dt):
|
|
traces = cls.env['mailing.trace'].sudo().create([
|
|
dict({'mass_mailing_id': mailing.id,
|
|
'model': record._name,
|
|
'res_id': record.id,
|
|
'trace_status': values.get('trace_status', 'bounce'),
|
|
# TDE FIXME: improve this with a mail-enabled heuristics
|
|
'email': record[fname],
|
|
'message_id': '<%5f@gilbert.boitempomils>' % randomized,
|
|
}, **values)
|
|
for record in records
|
|
])
|
|
return traces
|
|
|
|
|
|
class MassMailCommon(MailCommon, MassMailCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(MassMailCommon, cls).setUpClass()
|
|
|
|
cls.user_marketing = mail_new_test_user(
|
|
cls.env, login='user_marketing',
|
|
groups='base.group_user,base.group_partner_manager,mass_mailing.group_mass_mailing_user',
|
|
name='Martial Marketing', signature='--\nMartial')
|
|
|
|
cls.email_reply_to = 'MyCompany SomehowAlias <test.alias@test.mycompany.com>'
|
|
|
|
@classmethod
|
|
def _create_mailing_list(cls):
|
|
""" Shortcut to create mailing lists. Currently hardcoded, maybe evolve
|
|
in a near future. """
|
|
cls.mailing_list_1 = cls.env['mailing.list'].with_context(cls._test_context).create({
|
|
'name': 'List1',
|
|
'contact_ids': [
|
|
(0, 0, {'name': 'Déboulonneur', 'email': 'fleurus@example.com'}),
|
|
(0, 0, {'name': 'Gorramts', 'email': 'gorramts@example.com'}),
|
|
(0, 0, {'name': 'Ybrant', 'email': 'ybrant@example.com'}),
|
|
]
|
|
})
|
|
cls.mailing_list_2 = cls.env['mailing.list'].with_context(cls._test_context).create({
|
|
'name': 'List2',
|
|
'contact_ids': [
|
|
(0, 0, {'name': 'Gilberte', 'email': 'gilberte@example.com'}),
|
|
(0, 0, {'name': 'Gilberte En Mieux', 'email': 'gilberte@example.com'}),
|
|
(0, 0, {'name': 'Norbert', 'email': 'norbert@example.com'}),
|
|
(0, 0, {'name': 'Ybrant', 'email': 'ybrant@example.com'}),
|
|
]
|
|
})
|
|
|
|
@classmethod
|
|
def _create_mailing_list_of_x_contacts(cls, contacts_nbr):
|
|
""" Shortcut to create a mailing list that contains a defined number
|
|
of contacts. """
|
|
return cls.env['mailing.list'].with_context(cls._test_context).create({
|
|
'name': 'Test List',
|
|
'contact_ids': [
|
|
(0, 0, {'name': 'Contact %s' % i, 'email': 'contact%s@example.com' % i})
|
|
for i in range(contacts_nbr)
|
|
],
|
|
})
|