pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
eef262abf491ce71362761f0a94dea9e31d4b156
odoo_source/addons/mass_mailing/tests/common.py
T
std-odoo 97ca45e4f3 [IMP] mail, mass_mailing: store the bounce email and allow the user to read it 
Purpose
=======
The bounce emails aren't stored in Odoo, which can complicate the
debugging of the email sending.

Now, we store this bounce email, and we allow the user to read it from
the interface, so he can easily find the issue when an email sending
fail.

Specifications
==============
The bounce email is stored on the mail notification for standard emails
sending, and on the mailing traces when using mass mailing.

For some email providers (e.g. Yahoo), the "Final-Recipient" header is
not present. Normally, it allows us to retrieve the original recipient
of the email which bounced and then the partner. So if this header is
not there in a bounce email, we take the first recipient of the parent
<mail.message>.

Change the way that we parse the email body, for the bounce email.
For most email providers, the first mail body is the one that contains
the error and the next one contains the parent email body. So, the
current logic might ignore this body for Outlook and Yahoo.

Task-2116296

closes odoo/odoo#105923

Related: odoo/enterprise#34051
Signed-off-by: Warnon Aurélien (awa) <awa@odoo.com>
2023-03-06 17:47:29 +01:00

285 lines
13 KiB
Python
Raw Blame History

# -*- 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 'failure_reason' in recipient_info:
self.assertEqual(recipient_trace.failure_reason, recipient_info['failure_reason'])
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(),
'body': 'This is the bounce email',
}
if bounce_base_values:
parsed_bounce_values.update(bounce_base_values)
parsed_bounce_values.update({
'bounced_email': trace.email,
'bounced_msg_ids': [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)
],
})
Reference in New Issue View Git Blame Copy Permalink