Initial squashed commit

This commit is contained in:
hoangvv
2026-05-12 18:58:14 +07:00
commit 0b5c2ce91a
54896 changed files with 28050323 additions and 0 deletions
+6
View File
@@ -0,0 +1,6 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import common
from . import test_sms_composer
from . import test_sms_template
+369
View File
@@ -0,0 +1,369 @@
# -*- coding: utf-8 -*-
from contextlib import contextmanager
from freezegun import freeze_time
from unittest.mock import patch
from odoo import exceptions, tools
from odoo.addons.mail.tests.common import MailCommon
from odoo.addons.phone_validation.tools import phone_validation
from odoo.addons.sms.models.sms_sms import SmsSms
from odoo.addons.sms.tools.sms_api import SmsApi
from odoo.tests import common
class MockSMS(common.HttpCase):
def tearDown(self):
super(MockSMS, self).tearDown()
self._clear_sms_sent()
# ------------------------------------------------------------
# UTILITY MOCKS
# ------------------------------------------------------------
@contextmanager
def mock_datetime_and_now(self, mock_dt):
""" Used when synchronization date (using env.cr.now()) is important
in addition to standard datetime mocks. Used mainly to detect sync
issues. """
with freeze_time(mock_dt), \
patch.object(self.env.cr, 'now', lambda: mock_dt):
yield
# ------------------------------------------------------------
# GATEWAY MOCK
# ------------------------------------------------------------
@contextmanager
def mockSMSGateway(self, sms_allow_unlink=False, sim_error=None, nbr_t_error=None, moderated=False, force_delivered=False):
self._clear_sms_sent()
sms_create_origin = SmsSms.create
sms_send_origin = SmsSms._send
def _contact_iap(local_endpoint, params):
# mock single sms sending
if local_endpoint == '/iap/message_send':
self._sms += [{
'number': number,
'body': params['message'],
} for number in params['numbers']]
return True # send_message v0 API returns always True
# mock batch sending
if local_endpoint == '/iap/sms/2/send':
result = []
for to_send in params['messages']:
res = {'res_id': to_send['res_id'], 'state': 'delivered' if force_delivered else 'success', 'credit': 1}
error = sim_error or (nbr_t_error and nbr_t_error.get(to_send['number']))
if error and error == 'credit':
res.update(credit=0, state='insufficient_credit')
elif error and error in {'wrong_number_format', 'unregistered', 'server_error'}:
res.update(state=error)
elif error and error == 'jsonrpc_exception':
raise exceptions.AccessError(
'The url that this service requested returned an error. Please contact the author of the app. The url it tried to contact was ' + local_endpoint
)
result.append(res)
if res['state'] == 'success' or res['state'] == 'delivered':
self._sms.append({
'number': to_send['number'],
'body': to_send['content'],
})
return result
elif local_endpoint == '/api/sms/3/send':
result = []
for message in params['messages']:
for number in message["numbers"]:
error = sim_error or (nbr_t_error and nbr_t_error.get(number['number']))
if error == 'jsonrpc_exception':
raise exceptions.AccessError(
'The url that this service requested returned an error. '
'Please contact the author of the app. '
'The url it tried to contact was ' + local_endpoint
)
elif error == 'credit':
error = 'insufficient_credit'
res = {
'uuid': number['uuid'],
'state': error or ('delivered' if force_delivered else 'success' if not moderated else 'processing'),
'credit': 1,
}
if error:
# credit is only given if the amount is known
res.update(credit=0)
else:
self._sms.append({
'number': number['number'],
'body': message['content'],
'uuid': number['uuid'],
})
result.append(res)
return result
def _sms_sms_create(model, *args, **kwargs):
res = sms_create_origin(model, *args, **kwargs)
self._new_sms += res.sudo()
return res
def _sms_sms_send(records, unlink_failed=False, unlink_sent=True, raise_exception=False):
if sms_allow_unlink:
return sms_send_origin(records, unlink_failed=unlink_failed, unlink_sent=unlink_sent, raise_exception=raise_exception)
return sms_send_origin(records, unlink_failed=False, unlink_sent=False, raise_exception=raise_exception)
try:
with patch.object(SmsApi, '_contact_iap', side_effect=_contact_iap) as _sms_api_contact_iap_mock, \
patch.object(SmsSms, 'create', autospec=True, wraps=SmsSms, side_effect=_sms_sms_create) as sms_create, \
patch.object(SmsSms, '_send', autospec=True, wraps=SmsSms, side_effect=_sms_sms_send):
self._sms_api_contact_iap_mock = _sms_api_contact_iap_mock
self._mock_sms_create = sms_create
yield
finally:
pass
def _clear_sms_sent(self):
self._sms = []
self._new_sms = self.env['sms.sms'].sudo()
def _clear_outgoing_sms(self):
""" As SMS gateway mock keeps SMS, we may need to remove them manually
if there are several tests in the same tx. """
self.env['sms.sms'].sudo().search([('state', '=', 'outgoing')]).unlink()
class SMSCase(MockSMS):
""" Main test class to use when testing SMS integrations. Contains helpers and tools related
to notification sent by SMS. """
@classmethod
def setUpClass(cls):
super().setUpClass()
# This is called to make sure that an iap_account for sms already exists or if not is created.
cls.env['iap.account'].get('sms')
def _find_sms_sent(self, partner, number):
if number is None and partner:
number = partner._phone_format()
sent_sms = next((sms for sms in self._sms if sms['number'] == number), None)
if not sent_sms:
debug_info = '\n'.join(
f"To {sms['number']}"
for sms in self._sms
)
raise AssertionError(f'sent sms not found for {partner} (number: {number})\n{debug_info}')
return sent_sms
def _find_sms_sms(self, partner, number, status, content=None):
if number is None and partner:
number = partner._phone_format()
domain = [('id', 'in', self._new_sms.ids),
('partner_id', '=', partner.id),
('number', '=', number)]
if status:
domain += [('state', '=', status)]
sms = self.env['sms.sms'].sudo().search(domain)
if len(sms) > 1 and content:
sms = sms.filtered(lambda s: content in (s.body or ""))
if not sms:
debug_info = '\n'.join(
f"To {sms.number} ({sms.partner_id}) / state {sms.state}"
for sms in self._new_sms
)
raise AssertionError(
f'sms.sms not found for {partner} (number: {number} / status {status})\n--MOCKED DATA\n{debug_info}'
)
if len(sms) > 1:
raise NotImplementedError(
f'Found {len(sms)} sms.sms for {partner} (number: {number} / status {status})'
)
return sms
def assertSMSIapSent(self, numbers, content=None):
""" Check sent SMS (to IAP, but other providers like twilio should be
mocked to fill up 'self._sms', allowing tests to pass). Order is not
checked. Each number should have received the same content. Useful to
check batch sending.
:param numbers: list of numbers;
:param content: content to check for each number;
"""
for number in numbers:
sent_sms = next((sms for sms in self._sms if sms['number'] == number), None)
self.assertTrue(bool(sent_sms), 'Number %s not found in %s' % (number, repr([s['number'] for s in self._sms])))
if content is not None:
self.assertIn(content, sent_sms['body'])
def assertSMS(self, partner, number, status, failure_type=None,
content=None, fields_values=None):
""" Find a ``sms.sms`` record, based on given partner, number and status.
:param partner: optional partner, used to find a ``sms.sms`` and a number
if not given;
:param number: optional number, used to find a ``sms.sms``, notably if
partner is not given;
:param failure_type: check failure type if SMS is not sent or outgoing;
:param content: if given, should be contained in sms body;
:param fields_values: optional values allowing to check directly some
values on ``sms.sms`` record;
"""
sms_sms = self._find_sms_sms(partner, number, status, content=content)
if failure_type:
self.assertEqual(sms_sms.failure_type, failure_type)
if content is not None:
self.assertIn(content, (sms_sms.body or ""))
for fname, fvalue in (fields_values or {}).items():
self.assertEqual(
sms_sms[fname], fvalue,
'SMS: expected %s for %s, got %s' % (fvalue, fname, sms_sms[fname]))
if status == 'pending':
self.assertSMSIapSent([sms_sms.number], content=content)
def assertSMSCanceled(self, partner, number, failure_type, content=None, fields_values=None):
""" Check canceled SMS. Search is done for a pair partner / number where
partner can be an empty recordset. """
self.assertSMS(partner, number, 'canceled', failure_type=failure_type, content=content, fields_values=fields_values)
def assertSMSFailed(self, partner, number, failure_type, content=None, fields_values=None):
""" Check failed SMS. Search is done for a pair partner / number where
partner can be an empty recordset. """
self.assertSMS(partner, number, 'error', failure_type=failure_type, content=content, fields_values=fields_values)
def assertSMSOutgoing(self, partner, number, content=None, fields_values=None):
""" Check outgoing SMS. Search is done for a pair partner / number where
partner can be an empty recordset. """
self.assertSMS(partner, number, 'outgoing', content=content, fields_values=fields_values)
def assertNoSMSNotification(self, messages=None):
base_domain = [('notification_type', '=', 'sms')]
if messages is not None:
base_domain += [('mail_message_id', 'in', messages.ids)]
self.assertEqual(self.env['mail.notification'].search(base_domain), self.env['mail.notification'])
self.assertEqual(self._sms, [])
def assertSMSNotification(self, recipients_info, content, messages=None, check_sms=True, sent_unlink=False,
mail_message_values=None):
""" Check content of notifications and sms.
:param recipients_info: list[{
'partner': res.partner record (may be empty),
'number': number used for notification (may be empty, computed based on partner),
'state': ready / pending / sent / exception / canceled (pending by default),
'failure_type': optional: sms_number_missing / sms_number_format / sms_credit / sms_server
}, { ... }]
:param content: SMS content
:param mail_message_values: dictionary of expected mail message fields values
"""
partners = self.env['res.partner'].concat(*list(p['partner'] for p in recipients_info if p.get('partner')))
numbers = [p['number'] for p in recipients_info if p.get('number')]
# special case of void notifications: check for False / False notifications
if not partners and not numbers:
numbers = [False]
base_domain = [
'|', ('res_partner_id', 'in', partners.ids),
'&', ('res_partner_id', '=', False), ('sms_number', 'in', numbers),
('notification_type', '=', 'sms')
]
if messages is not None:
base_domain += [('mail_message_id', 'in', messages.ids)]
notifications = self.env['mail.notification'].search(base_domain)
self.assertEqual(notifications.mapped('res_partner_id'), partners)
for recipient_info in recipients_info:
# sanity check
extra_keys = recipient_info.keys() - {
# notification
'failure_reason',
'failure_type',
'state',
# sms
'sms_fields_values',
# recipient
'number',
'partner',
'recipient_check_sms',
}
if extra_keys:
raise ValueError(f'Unsupported values: {extra_keys}')
partner = recipient_info.get('partner', self.env['res.partner'])
number = recipient_info.get('number')
state = recipient_info.get('state', 'pending')
if number is None and partner:
number = partner._phone_format()
notif = notifications.filtered(lambda n: n.res_partner_id == partner and n.sms_number == number and n.notification_status == state)
debug_info = ''
if not notif:
debug_info = '\n'.join(
f'To: {notif.sms_number} ({notif.res_partner_id}) - (State: {notif.notification_status})'
for notif in notifications
)
self.assertTrue(notif, 'SMS: not found notification for %s (number: %s, state: %s)\n%s' % (partner, number, state, debug_info))
self.assertEqual(notif.author_id, notif.mail_message_id.author_id, 'SMS: Message and notification should have the same author')
for field_name, expected_value in (mail_message_values or {}).items():
self.assertEqual(notif.mail_message_id[field_name], expected_value)
if 'failure_reason' in recipient_info:
self.assertEqual(notif.failure_reason, recipient_info['failure_reason'])
if state not in {'process', 'sent', 'ready', 'canceled', 'pending'}:
self.assertEqual(notif.failure_type, recipient_info['failure_type'])
if recipient_info.get('recipient_check_sms', check_sms):
fields_values = recipient_info.get('sms_fields_values') or {}
if state in {'process', 'pending', 'sent'}:
if sent_unlink:
self.assertSMSIapSent([number], content=content)
else:
self.assertSMS(partner, number, state, content=content, fields_values=fields_values)
elif state == 'ready':
self.assertSMS(partner, number, 'outgoing', content=content, fields_values=fields_values)
elif state == 'exception':
self.assertSMS(partner, number, 'error', failure_type=recipient_info['failure_type'], content=content, fields_values=fields_values)
elif state == 'canceled':
self.assertSMS(partner, number, 'canceled', failure_type=recipient_info['failure_type'], content=content, fields_values=fields_values)
else:
raise NotImplementedError('Not implemented')
if messages is not None:
sanitize_tags = {**tools.mail.SANITIZE_TAGS}
sanitize_tags['remove_tags'] = [*sanitize_tags['remove_tags'] + ['a']]
with patch('odoo.tools.mail.SANITIZE_TAGS', sanitize_tags):
for message in messages:
self.assertEqual(content, tools.html2plaintext(tools.html_sanitize(message.body)).rstrip('\n'))
def assertSMSLogged(self, records, body):
for record in records:
message = record.message_ids[-1]
self.assertEqual(message.subtype_id, self.env.ref('mail.mt_note'))
self.assertEqual(message.message_type, 'sms')
self.assertEqual(tools.html2plaintext(message.body).rstrip('\n'), body)
class SMSCommon(MailCommon, SMSCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
# some numbers for testing
cls.random_numbers_str = '+32456998877, 0456665544'
cls.random_numbers = cls.random_numbers_str.split(', ')
cls.random_numbers_san = [phone_validation.phone_format(number, 'BE', '32', force_format='E164') for number in cls.random_numbers]
cls.test_numbers = ['+32456010203', '0456 04 05 06', '0032456070809']
cls.test_numbers_san = [phone_validation.phone_format(number, 'BE', '32', force_format='E164') for number in cls.test_numbers]
# some numbers for mass testing
cls.mass_numbers = ['04561%s2%s3%s' % (x, x, x) for x in range(0, 10)]
cls.mass_numbers_san = [phone_validation.phone_format(number, 'BE', '32', force_format='E164') for number in cls.mass_numbers]
@classmethod
def _create_sms_template(cls, model, body=False):
return cls.env['sms.template'].create({
'name': 'Test Template',
'model_id': cls.env['ir.model']._get(model).id,
'body': body if body else 'Dear {{ object.display_name }} this is an SMS.'
})
def _make_webhook_jsonrpc_request(self, statuses):
return self.make_jsonrpc_request('/sms/status', {'message_statuses': statuses})
+84
View File
@@ -0,0 +1,84 @@
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from unittest.mock import patch
from odoo.addons.sms.models.mail_thread import MailThread
from odoo.addons.sms.tests.common import SMSCommon, SMSCase
from odoo.tests import tagged
@tagged('at_install')
class TestSMSComposerComment(SMSCommon, SMSCase):
""" Test behaviors that are overridden when other modules
are installed (e.g., mass_mailing). In these cases,
test_mail_sms or test_mail_full should be used."""
def test_message_post_sms_vs_notification(self):
"""Check that the conversion of html to plain text does remove links
This is necessary when an SMS is sent from message_post with sms type
and not from _message_sms. In this case, it can be expected to receive html
that should be interpreted as such instead of escaped before being sent.
"""
cases = [
(
'Hello there, check this awesome <b>app</b> I found:<br/>https://odoo.com', # not a `a` link in source
'<p>Hello there, check this awesome &lt;b&gt;app&lt;/b&gt; I found:&lt;br/&gt;<a href="https://odoo.com" target="_blank" rel="noreferrer noopener">https://odoo.com</a></p>',
'Hello there, check this awesome <b>app</b> I found:<br/>https://odoo.com'
), (
'Hello there, check this awesome <b>app</b> I found:<br/><a href="https://odoo.com">Here</a>', # a link
'<p>Hello there, check this awesome &lt;b&gt;app&lt;/b&gt; I found:&lt;br/&gt;&lt;a href="<a href="https://odoo.com" target="_blank" rel="noreferrer noopener">https://odoo.com</a>"&gt;Here&lt;/a&gt;</p>',
'Hello there, check this awesome <b>app</b> I found:<br/><a href="https://odoo.com">Here</a>' # keep all information
)
]
for message_content, expected_notification_content, expected_sms_content in cases:
with self.subTest(message_content=message_content):
with self.with_user('admin'), self.mockSMSGateway():
message = self.env.user.partner_id.message_post(
body=message_content, message_type='sms', sms_numbers=['+3215228817386'])
self.assertSMSNotification(
[{'number': '+3215228817386'}], expected_sms_content, message,
mail_message_values={"body": expected_notification_content},
)
def test_message_sms_body_sms_vs_notification(self):
"""Check that the rendering of the sms notification is identical to the sms.
The only expected difference is that links are converted to be clickable.
The test verifies that MailThread._message_sms() works as expected."""
# Cases are formatted as sms text, expected notification body
cases = [
(
"Hello there, check this awesome app I found:\nhttps://odoo.com",
'<p>Hello there, check this awesome app I found:<br>'
'<a href="https://odoo.com" target="_blank" rel="noreferrer noopener">https://odoo.com</a></p>',
), (
"Hello there, check this awesome <b>app</b> I found:\nhttps://odoo.com",
# b is kept as is in notification, but link is still added as well
'<p>Hello there, check this awesome &lt;b&gt;app&lt;/b&gt; I found:<br>'
'<a href="https://odoo.com" target="_blank" rel="noreferrer noopener">https://odoo.com</a></p>',
),
(
# Here, we check that the sms sent is the sms written.
"Hello there, check this awesome <b>app</b> I found:\n*https://odoo.com*",
'<p>Hello there, check this awesome &lt;b&gt;app&lt;/b&gt; I found:<br>'
'*<a href="https://odoo.com" target="_blank" rel="noreferrer noopener">https://odoo.com</a>*</p>',
),
]
for sms_content, expected_notification_content in cases:
with self.subTest(sms_content=sms_content):
with self.with_user('admin'):
composer = self.env['sms.composer'].with_context(
active_model='res.partner', active_id=self.partner_employee).create({'body': sms_content})
_message_sms_patch = patch.object(
MailThread, '_message_sms', autospec=True, side_effect=MailThread._message_sms)
with self.mockSMSGateway(), _message_sms_patch as _patched_message_sms:
messages = composer._action_send_sms()
_patched_message_sms.assert_called() # make sure we're testing `_message_sms` too
self.assertSMSNotification(
[{'partner': self.partner_employee}], sms_content, messages,
mail_message_values={"body": expected_notification_content},
)
+138
View File
@@ -0,0 +1,138 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from markupsafe import Markup
from odoo.tests.common import TransactionCase, users
from odoo.addons.mail.tests.common import mail_new_test_user
from odoo.exceptions import AccessError
from odoo.tests import tagged
from odoo.tools import mute_logger, convert_file
@tagged('post_install', '-at_install')
class TestSmsTemplateAccessRights(TransactionCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.user_admin = mail_new_test_user(cls.env, login='user_system', groups='base.group_user,base.group_system')
cls.basic_user = mail_new_test_user(cls.env, login='user_employee', groups='base.group_user')
sms_enabled_models = cls.env['ir.model'].search([('is_mail_thread', '=', True), ('transient', '=', False)])
vals = []
for model in sms_enabled_models:
vals.append({
'name': 'SMS Template ' + model.name,
'body': 'Body Test',
'model_id': model.id,
})
cls.sms_templates = cls.env['sms.template'].create(vals)
cls.sms_dynamic_template = cls.env['sms.template'].sudo().create({
'body': '{{ object.name }}',
'model_id': cls.env['ir.model'].sudo().search([('model', '=', 'res.partner')]).id,
})
cls.partner = cls.env['res.partner'].create({'name': 'Test Partner'})
@users('user_employee')
@mute_logger('odoo.models.unlink')
def test_access_rights_user(self):
# Check if a member of group_user can only read on sms.template
for sms_template in self.env['sms.template'].browse(self.sms_templates.ids):
self.assertTrue(bool(sms_template.name))
with self.assertRaises(AccessError):
sms_template.write({'name': 'Update Template'})
with self.assertRaises(AccessError):
self.env['sms.template'].create({
'name': 'New SMS Template ' + sms_template.model_id.name,
'body': 'Body Test',
'model_id': sms_template.model_id.id,
})
with self.assertRaises(AccessError):
sms_template.unlink()
@users('user_system')
@mute_logger('odoo.models.unlink', 'odoo.addons.base.models.ir_model')
def test_access_rights_system(self):
admin = self.env.ref('base.user_admin')
for sms_template in self.env['sms.template'].browse(self.sms_templates.ids):
self.assertTrue(bool(sms_template.name))
sms_template.write({'body': 'New body from admin'})
self.env['sms.template'].create({
'name': 'New SMS Template ' + sms_template.model_id.name,
'body': 'Body Test',
'model_id': sms_template.model_id.id,
})
# check admin is allowed to read all templates since he can be a member of
# other groups applying restrictions based on the model
self.assertTrue(bool(self.env['sms.template'].with_user(admin).browse(sms_template.ids).name))
sms_template.unlink()
@users('user_employee')
def test_sms_template_rendering_restricted(self):
self.env['ir.config_parameter'].sudo().set_param('mail.restrict.template.rendering', True)
self.basic_user.groups_id -= self.env.ref('mail.group_mail_template_editor')
sms_composer = self.env['sms.composer'].create({
'composition_mode': 'comment',
'template_id': self.sms_dynamic_template.id,
'res_id': self.partner.id,
'res_model': 'res.partner',
})
self.assertEqual(sms_composer.body, self.partner.name, 'Simple user should be able to render SMS template')
sms_composer.composition_mode = 'mass'
self.assertEqual(sms_composer.body, '{{ object.name }}', 'In mass mode, we should not render the template')
body = sms_composer._prepare_body_values(self.partner)[self.partner.id]
self.assertEqual(body, self.partner.name, 'In mass mode, if the user did not change the body, he should be able to render it')
sms_composer.body = 'New body: {{ 4 + 9 }}'
with self.assertRaises(AccessError, msg='User should not be able to write new inline_template code'):
sms_composer._prepare_body_values(self.partner)
@users('user_system')
def test_sms_template_rendering_unrestricted(self):
self.env['ir.config_parameter'].sudo().set_param('mail.restrict.template.rendering', True)
sms_composer = self.env['sms.composer'].create({
'composition_mode': 'comment',
'template_id': self.sms_dynamic_template.id,
'res_id': self.partner.id,
'res_model': 'res.partner',
})
body = sms_composer._prepare_body_values(self.partner)[self.partner.id]
self.assertIn(self.partner.name, body, 'Template Editor should be able to write new Jinja code')
@tagged('post_install', '-at_install')
class TestSMSTemplateReset(TransactionCase):
def _load(self, module, filepath):
# pylint: disable=no-value-for-parameter
convert_file(self.env, module='sms',
filename=filepath,
idref={}, mode='init', noupdate=False, kind='test')
def test_sms_template_reset(self):
self._load('sms', 'tests/test_sms_template.xml')
sms_template = self.env.ref('sms.sms_template_test').with_context(lang=self.env.user.lang)
sms_template.write({
'body': '<div>Hello</div>',
'name': 'SMS: SMS Template',
})
context = {'default_template_ids': sms_template.ids}
sms_template_reset = self.env['sms.template.reset'].with_context(context).create({})
reset_action = sms_template_reset.reset_template()
self.assertTrue(reset_action)
self.assertEqual(sms_template.body.strip(), Markup('<div>Hello Odoo</div>'))
# Name is not there in the data file template, so it should be set to False
self.assertFalse(sms_template.name, "Name should be set to False")
+9
View File
@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="utf-8"?>
<odoo>
<record id="sms_template_test" model="sms.template">
<field name="model_id" ref="base.model_res_partner"/>
<field name="body" type="html">
<div>Hello Odoo</div>
</field>
</record>
</odoo>