Thibault Delavallée
e5338146e7
RATIONALE
Channel model is a mail.thread enabled model behaving strangely with followers,
notifications and discuss. Its code should however be simplified to be more
self contained and avoid unwanted side effects on other models.
PURPOSE
Remove channel ability to follow records as it mainly adds noise without a lot
of added value. Simplify channel notification flow by using directly members
and not a delegation through a channel self-following trick. Remove followers
being channels and posting with added listeners being channels.
SPECIFICATIONS
In this commit we force messages to belong to a single document using
``model`` / ``res_id`` pair. It is not possible anymore to link a message
to channels using ``channel_ids``. A message belongs to a document and
is displayed in that document's chatter.
This change implies modifying a lot of domains, notably in chatter. Indeed
discuss for channels does not use ``('channel_ids', 'in', [3])`` domains.
They now use ``('model', '=', 'mail.channel'), ('res_id', 'in', [3])`` like
other documents fetching their messages.
This commit also removes ``channel_message_ids`` field on ``mail.channel``
model. As channels are now considered as standard documents they will use
``message_ids`` field like all other documents. Linking a channel on a message
is possible only as a link in message from now on. It is not possible to push
it into a channel anymore (no more listener channels, no more channel link).
Finally a global cleaning also linked to all previous commits is done.
LINKS
Task ID-2070632 (main task)
Task ID-2419762 (followup task)
COM PR odoo/odoo#62859
ENT PR odoo/enterprise#15172
UPG PR odoo/upgrade#2005
357 lines
19 KiB
Python
357 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
|
|
|
import base64
|
|
|
|
from unittest.mock import patch
|
|
|
|
from odoo.addons.test_mail.data.test_mail_data import MAIL_TEMPLATE_PLAINTEXT
|
|
from odoo.addons.test_mail.models.test_mail_models import MailTestSimple
|
|
from odoo.addons.test_mail.tests.common import TestMailCommon, TestRecipients
|
|
from odoo.api import call_kw
|
|
from odoo.exceptions import AccessError
|
|
from odoo.tests import tagged
|
|
from odoo.tools import mute_logger, formataddr
|
|
|
|
|
|
@tagged('mail_post')
|
|
class TestMessagePost(TestMailCommon, TestRecipients):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(TestMessagePost, cls).setUpClass()
|
|
cls._create_portal_user()
|
|
cls.test_record = cls.env['mail.test.simple'].with_context(cls._test_context).create({'name': 'Test', 'email_from': 'ignasse@example.com'})
|
|
cls._reset_mail_context(cls.test_record)
|
|
cls.user_admin.write({'notification_type': 'email'})
|
|
|
|
# This method should be run inside a post_install class to ensure that all
|
|
# message_post overrides are tested.
|
|
def test_message_post_return(self):
|
|
test_channel = self.env['mail.channel'].create({
|
|
'name': 'Test',
|
|
})
|
|
# Use call_kw as shortcut to simulate a RPC call.
|
|
messageId = call_kw(self.env['mail.channel'], 'message_post', [test_channel.id], {'body': 'test'})
|
|
self.assertTrue(isinstance(messageId, int))
|
|
|
|
def test_notify_recipients_internals(self):
|
|
pdata = self._generate_notify_recipients(self.partner_1 | self.partner_employee)
|
|
msg_vals = {
|
|
'body': 'Message body',
|
|
'model': self.test_record._name,
|
|
'res_id': self.test_record.id,
|
|
'subject': 'Message subject',
|
|
}
|
|
link_vals = {
|
|
'token': 'token_val',
|
|
'access_token': 'access_token_val',
|
|
'auth_signup_token': 'auth_signup_token_val',
|
|
'auth_login': 'auth_login_val',
|
|
}
|
|
notify_msg_vals = dict(msg_vals, **link_vals)
|
|
classify_res = self.env[self.test_record._name]._notify_classify_recipients(pdata, 'My Custom Model Name', msg_vals=notify_msg_vals)
|
|
# find back information for each recipients
|
|
partner_info = next(item for item in classify_res if item['recipients'] == self.partner_1.ids)
|
|
emp_info = next(item for item in classify_res if item['recipients'] == self.partner_employee.ids)
|
|
|
|
# partner: no access button
|
|
self.assertFalse(partner_info['has_button_access'])
|
|
|
|
# employee: access button and link
|
|
self.assertTrue(emp_info['has_button_access'])
|
|
for param, value in link_vals.items():
|
|
self.assertIn('%s=%s' % (param, value), emp_info['button_access']['url'])
|
|
self.assertIn('model=%s' % self.test_record._name, emp_info['button_access']['url'])
|
|
self.assertIn('res_id=%s' % self.test_record.id, emp_info['button_access']['url'])
|
|
self.assertNotIn('body', emp_info['button_access']['url'])
|
|
self.assertNotIn('subject', emp_info['button_access']['url'])
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_needaction(self):
|
|
(self.user_employee | self.user_admin).write({'notification_type': 'inbox'})
|
|
with self.assertSinglePostNotifications([{'partner': self.partner_employee, 'type': 'inbox'}], {'content': 'Body'}):
|
|
self.test_record.message_post(
|
|
body='Body', message_type='comment', subtype_xmlid='mail.mt_comment',
|
|
partner_ids=[self.user_employee.partner_id.id])
|
|
|
|
self.test_record.message_subscribe([self.partner_1.id])
|
|
with self.assertSinglePostNotifications([
|
|
{'partner': self.partner_employee, 'type': 'inbox'},
|
|
{'partner': self.partner_1, 'type': 'email'}], {'content': 'NewBody'}):
|
|
self.test_record.message_post(
|
|
body='NewBody', message_type='comment', subtype_xmlid='mail.mt_comment',
|
|
partner_ids=[self.user_employee.partner_id.id])
|
|
|
|
with self.assertSinglePostNotifications([
|
|
{'partner': self.partner_1, 'type': 'email'},
|
|
{'partner': self.partner_portal, 'type': 'email'}], {'content': 'ToPortal'}):
|
|
self.test_record.message_post(
|
|
body='ToPortal', message_type='comment', subtype_xmlid='mail.mt_comment',
|
|
partner_ids=[self.partner_portal.id])
|
|
|
|
def test_post_inactive_follower(self):
|
|
# In some case odoobot is follower of a record.
|
|
# Even if it shouldn't be the case, we want to be sure that odoobot is not notified
|
|
(self.user_employee | self.user_admin).write({'notification_type': 'inbox'})
|
|
self.test_record._message_subscribe(self.user_employee.partner_id.ids)
|
|
with self.assertSinglePostNotifications([{'partner': self.partner_employee, 'type': 'inbox'}], {'content': 'Test'}):
|
|
self.test_record.message_post(
|
|
body='Test', message_type='comment', subtype_xmlid='mail.mt_comment')
|
|
|
|
self.user_employee.active = False
|
|
# at this point, partner is still active and would receive an email notification
|
|
self.user_employee.partner_id._write({'active': False})
|
|
with self.assertPostNotifications([{'content': 'Test', 'notif': []}]):
|
|
self.test_record.message_post(
|
|
body='Test', message_type='comment', subtype_xmlid='mail.mt_comment')
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_notifications(self):
|
|
_body, _subject = '<p>Test Body</p>', 'Test Subject'
|
|
|
|
# subscribe second employee to the group to test notifications
|
|
self.test_record.message_subscribe(partner_ids=[self.user_admin.partner_id.id])
|
|
|
|
with self.assertSinglePostNotifications([
|
|
{'partner': self.partner_1, 'type': 'email'},
|
|
{'partner': self.partner_2, 'type': 'email'},
|
|
{'partner': self.partner_admin, 'type': 'email'}], {'content': _body}, mail_unlink_sent=True):
|
|
msg = self.test_record.with_user(self.user_employee).message_post(
|
|
body=_body, subject=_subject,
|
|
message_type='comment', subtype_xmlid='mail.mt_comment',
|
|
partner_ids=[self.partner_1.id, self.partner_2.id]
|
|
)
|
|
|
|
# message content
|
|
self.assertEqual(msg.subject, _subject)
|
|
self.assertEqual(msg.body, _body)
|
|
self.assertEqual(msg.partner_ids, self.partner_1 | self.partner_2)
|
|
self.assertEqual(msg.notified_partner_ids, self.user_admin.partner_id | self.partner_1 | self.partner_2)
|
|
|
|
# notifications emails should have been deleted
|
|
self.assertFalse(self.env['mail.mail'].sudo().search([('mail_message_id', '=', msg.id)]),
|
|
'message_post: mail.mail notifications should have been auto-deleted')
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_notifications_keep_emails(self):
|
|
self.test_record.message_subscribe(partner_ids=[self.user_admin.partner_id.id])
|
|
|
|
msg = self.test_record.with_user(self.user_employee).message_post(
|
|
body='Test', subject='Test',
|
|
message_type='comment', subtype_xmlid='mail.mt_comment',
|
|
partner_ids=[self.partner_1.id, self.partner_2.id],
|
|
mail_auto_delete=False
|
|
)
|
|
|
|
# notifications emails should not have been deleted: one for customers, one for user
|
|
self.assertEqual(len(self.env['mail.mail'].sudo().search([('mail_message_id', '=', msg.id)])), 2)
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_notifications_emails_tweak(self):
|
|
pass
|
|
# we should check _notification_groups behavior, for emails and buttons
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_attachments(self):
|
|
_attachments = [
|
|
('List1', b'My first attachment'),
|
|
('List2', b'My second attachment')
|
|
]
|
|
_attach_1 = self.env['ir.attachment'].with_user(self.user_employee).create({
|
|
'name': 'Attach1',
|
|
'datas': 'bWlncmF0aW9uIHRlc3Q=',
|
|
'res_model': 'mail.compose.message', 'res_id': 0})
|
|
_attach_2 = self.env['ir.attachment'].with_user(self.user_employee).create({
|
|
'name': 'Attach2',
|
|
'datas': 'bWlncmF0aW9uIHRlc3Q=',
|
|
'res_model': 'mail.compose.message', 'res_id': 0})
|
|
|
|
with self.mock_mail_gateway():
|
|
msg = self.test_record.with_user(self.user_employee).message_post(
|
|
body='Test', subject='Test',
|
|
message_type='comment', subtype_xmlid='mail.mt_comment',
|
|
attachment_ids=[_attach_1.id, _attach_2.id],
|
|
partner_ids=[self.partner_1.id],
|
|
attachments=_attachments,
|
|
)
|
|
|
|
# message attachments
|
|
self.assertEqual(len(msg.attachment_ids), 4)
|
|
self.assertEqual(set(msg.attachment_ids.mapped('res_model')), set([self.test_record._name]))
|
|
self.assertEqual(set(msg.attachment_ids.mapped('res_id')), set([self.test_record.id]))
|
|
self.assertEqual(set([base64.b64decode(x) for x in msg.attachment_ids.mapped('datas')]),
|
|
set([b'migration test', _attachments[0][1], _attachments[1][1]]))
|
|
self.assertTrue(set([_attach_1.id, _attach_2.id]).issubset(msg.attachment_ids.ids),
|
|
'message_post: mail.message attachments duplicated')
|
|
|
|
# notification email attachments
|
|
self.assertSentEmail(self.user_employee.partner_id, [self.partner_1])
|
|
# self.assertEqual(len(self._mails), 1)
|
|
self.assertEqual(len(self._mails[0]['attachments']), 4)
|
|
self.assertIn(('List1', b'My first attachment', 'application/octet-stream'), self._mails[0]['attachments'])
|
|
self.assertIn(('List2', b'My second attachment', 'application/octet-stream'), self._mails[0]['attachments'])
|
|
self.assertIn(('Attach1', b'migration test', 'application/octet-stream'), self._mails[0]['attachments'])
|
|
self.assertIn(('Attach2', b'migration test', 'application/octet-stream'), self._mails[0]['attachments'])
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_answer(self):
|
|
with self.mock_mail_gateway():
|
|
parent_msg = self.test_record.with_user(self.user_employee).message_post(
|
|
body='<p>Test</p>', subject='Test Subject',
|
|
message_type='comment', subtype_xmlid='mail.mt_comment')
|
|
|
|
self.assertEqual(parent_msg.partner_ids, self.env['res.partner'])
|
|
self.assertNotSentEmail()
|
|
|
|
with self.assertPostNotifications([{'content': '<p>Test Answer</p>', 'notif': [{'partner': self.partner_1, 'type': 'email'}]}]):
|
|
msg = self.test_record.with_user(self.user_employee).message_post(
|
|
body='<p>Test Answer</p>',
|
|
message_type='comment', subtype_xmlid='mail.mt_comment',
|
|
partner_ids=[self.partner_1.id],
|
|
parent_id=parent_msg.id)
|
|
|
|
self.assertEqual(msg.parent_id.id, parent_msg.id)
|
|
self.assertEqual(msg.partner_ids, self.partner_1)
|
|
self.assertEqual(parent_msg.partner_ids, self.env['res.partner'])
|
|
|
|
# check notification emails: references
|
|
self.assertSentEmail(self.user_employee.partner_id, [self.partner_1], ref_content='openerp-%d-mail.test.simple' % self.test_record.id)
|
|
# self.assertTrue(all('openerp-%d-mail.test.simple' % self.test_record.id in m['references'] for m in self._mails))
|
|
|
|
new_msg = self.test_record.with_user(self.user_employee).message_post(
|
|
body='<p>Test Answer Bis</p>',
|
|
message_type='comment', subtype_xmlid='mail.mt_comment',
|
|
parent_id=msg.id)
|
|
|
|
self.assertEqual(new_msg.parent_id.id, parent_msg.id, 'message_post: flatten error')
|
|
self.assertEqual(new_msg.partner_ids, self.env['res.partner'])
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_portal_ok(self):
|
|
self.test_record.message_subscribe((self.partner_1 | self.user_employee.partner_id).ids)
|
|
|
|
with self.assertPostNotifications([{'content': '<p>Test</p>', 'notif': [
|
|
{'partner': self.partner_employee, 'type': 'inbox'},
|
|
{'partner': self.partner_1, 'type': 'email'}]}
|
|
]), patch.object(MailTestSimple, 'check_access_rights', return_value=True):
|
|
new_msg = self.test_record.with_user(self.user_portal).message_post(
|
|
body='<p>Test</p>', subject='Subject',
|
|
message_type='comment', subtype_xmlid='mail.mt_comment')
|
|
|
|
self.assertEqual(new_msg.sudo().notified_partner_ids, (self.partner_1 | self.user_employee.partner_id))
|
|
|
|
def test_post_portal_crash(self):
|
|
with self.assertRaises(AccessError):
|
|
self.test_record.with_user(self.user_portal).message_post(
|
|
body='<p>Test</p>', subject='Subject',
|
|
message_type='comment', subtype_xmlid='mail.mt_comment')
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail', 'odoo.addons.mail.models.mail_thread')
|
|
def test_post_internal(self):
|
|
self.test_record.message_subscribe([self.user_admin.partner_id.id])
|
|
msg = self.test_record.with_user(self.user_employee).message_post(
|
|
body='My Body', subject='My Subject',
|
|
message_type='comment', subtype_xmlid='mail.mt_note')
|
|
self.assertEqual(msg.partner_ids, self.env['res.partner'])
|
|
self.assertEqual(msg.notified_partner_ids, self.env['res.partner'])
|
|
|
|
self.format_and_process(
|
|
MAIL_TEMPLATE_PLAINTEXT, self.user_admin.email, 'not_my_businesss@example.com',
|
|
msg_id='<1198923581.41972151344608186800.JavaMail.diff1@agrolait.com>',
|
|
extra='In-Reply-To:\r\n\t%s\n' % msg.message_id,
|
|
target_model='mail.test.simple')
|
|
reply = self.test_record.message_ids - msg
|
|
self.assertTrue(reply)
|
|
self.assertEqual(reply.subtype_id, self.env.ref('mail.mt_note'))
|
|
self.assertEqual(reply.notified_partner_ids, self.user_employee.partner_id)
|
|
self.assertEqual(reply.parent_id, msg)
|
|
|
|
def test_post_log(self):
|
|
new_note = self.test_record.with_user(self.user_employee)._message_log(
|
|
body='<p>Labrador</p>',
|
|
)
|
|
|
|
self.assertEqual(new_note.subtype_id, self.env.ref('mail.mt_note'))
|
|
self.assertEqual(new_note.body, '<p>Labrador</p>')
|
|
self.assertEqual(new_note.author_id, self.user_employee.partner_id)
|
|
self.assertEqual(new_note.email_from, formataddr((self.user_employee.name, self.user_employee.email)))
|
|
self.assertEqual(new_note.notified_partner_ids, self.env['res.partner'])
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_notify(self):
|
|
self.user_employee.write({'notification_type': 'inbox'})
|
|
|
|
with self.mock_mail_gateway():
|
|
new_notification = self.test_record.message_notify(
|
|
subject='This should be a subject',
|
|
body='<p>You have received a notification</p>',
|
|
partner_ids=[self.partner_1.id, self.partner_admin.id, self.user_employee.partner_id.id],
|
|
)
|
|
|
|
self.assertEqual(new_notification.subtype_id, self.env.ref('mail.mt_note'))
|
|
self.assertEqual(new_notification.message_type, 'user_notification')
|
|
self.assertEqual(new_notification.body, '<p>You have received a notification</p>')
|
|
self.assertEqual(new_notification.author_id, self.env.user.partner_id)
|
|
self.assertEqual(new_notification.email_from, formataddr((self.env.user.name, self.env.user.email)))
|
|
self.assertEqual(new_notification.notified_partner_ids, self.partner_1 | self.user_employee.partner_id | self.partner_admin)
|
|
self.assertNotIn(new_notification, self.test_record.message_ids)
|
|
|
|
admin_mails = [x for x in self._mails if self.partner_admin.name in x.get('email_to')[0]]
|
|
self.assertEqual(len(admin_mails), 1, 'There should be exactly one email sent to admin')
|
|
admin_mail = admin_mails[0].get('body')
|
|
admin_access_link = admin_mail[admin_mail.index('model='):admin_mail.index('/>') - 1] if 'model=' in admin_mail else None
|
|
|
|
self.assertIsNotNone(admin_access_link, 'The email sent to admin should contain an access link')
|
|
self.assertIn('model=%s' % self.test_record._name, admin_access_link, 'The access link should contain a valid model argument')
|
|
self.assertIn('res_id=%d' % self.test_record.id, admin_access_link, 'The access link should contain a valid res_id argument')
|
|
|
|
partner_mails = [x for x in self._mails if self.partner_1.name in x.get('email_to')[0]]
|
|
self.assertEqual(len(partner_mails), 1, 'There should be exactly one email sent to partner')
|
|
partner_mail = partner_mails[0].get('body')
|
|
self.assertNotIn('/mail/view?model=', partner_mail, 'The email sent to admin should not contain an access link')
|
|
# todo xdo add test message_notify on thread with followers and stuff
|
|
|
|
@mute_logger('odoo.addons.mail.models.mail_mail')
|
|
def test_post_post_w_template(self):
|
|
test_record = self.env['mail.test.simple'].with_context(self._test_context).create({'name': 'Test', 'email_from': 'ignasse@example.com'})
|
|
self.user_employee.write({
|
|
'groups_id': [(4, self.env.ref('base.group_partner_manager').id)],
|
|
})
|
|
_attachments = [{
|
|
'name': 'first.txt',
|
|
'datas': base64.b64encode(b'My first attachment'),
|
|
'res_model': 'res.partner',
|
|
'res_id': self.user_admin.partner_id.id
|
|
}, {
|
|
'name': 'second.txt',
|
|
'datas': base64.b64encode(b'My second attachment'),
|
|
'res_model': 'res.partner',
|
|
'res_id': self.user_admin.partner_id.id
|
|
}]
|
|
email_1 = 'test1@example.com'
|
|
email_2 = 'test2@example.com'
|
|
email_3 = self.partner_1.email
|
|
self._create_template('mail.test.simple', {
|
|
'attachment_ids': [(0, 0, _attachments[0]), (0, 0, _attachments[1])],
|
|
'partner_to': '%s,%s' % (self.partner_2.id, self.user_admin.partner_id.id),
|
|
'email_to': '%s, %s' % (email_1, email_2),
|
|
'email_cc': '%s' % email_3,
|
|
})
|
|
# admin should receive emails
|
|
self.user_admin.write({'notification_type': 'email'})
|
|
# Force the attachments of the template to be in the natural order.
|
|
self.email_template.invalidate_cache(['attachment_ids'], ids=self.email_template.ids)
|
|
|
|
with self.mock_mail_gateway():
|
|
test_record.with_user(self.user_employee).message_post_with_template(self.email_template.id, composition_mode='comment')
|
|
|
|
new_partners = self.env['res.partner'].search([('email', 'in', [email_1, email_2])])
|
|
for r in [self.partner_1, self.partner_2, new_partners[0], new_partners[1], self.partner_admin]:
|
|
self.assertSentEmail(
|
|
self.user_employee.partner_id,
|
|
[r],
|
|
subject='About %s' % test_record.name,
|
|
body_content=test_record.name,
|
|
attachments=[('first.txt', b'My first attachment', 'text/plain'), ('second.txt', b'My second attachment', 'text/plain')])
|