initial commit 2

3 files changed, 345 insertions, 0 deletions

diff --git a/addons/sms/tests/__init__.py b/addons/sms/tests/__init__.py
new file mode 100644
index 00000000..609586be
--- /dev/null
+++ b/addons/sms/tests/__init__.py
@@ -0,0 +1,5 @@
+# -*- coding: utf-8 -*-
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+
+from . import common
+from . import test_sms_template
diff --git a/addons/sms/tests/common.py b/addons/sms/tests/common.py
new file mode 100644
index 00000000..03b31dd8
--- /dev/null
+++ b/addons/sms/tests/common.py
@@ -0,0 +1,277 @@
+# -*- coding: utf-8 -*-
+
+from contextlib import contextmanager
+from unittest.mock import patch
+
+from odoo import exceptions, tools
+from odoo.tests import common
+from odoo.addons.mail.tests.common import MailCommon
+from odoo.addons.phone_validation.tools import phone_validation
+from odoo.addons.sms.models.sms_api import SmsApi
+from odoo.addons.sms.models.sms_sms import SmsSms
+
+
+class MockSMS(common.BaseCase):
+
+    def tearDown(self):
+        super(MockSMS, self).tearDown()
+        self._clear_sms_sent()
+
+    @contextmanager
+    def mockSMSGateway(self, sms_allow_unlink=False, sim_error=None, nbr_t_error=None):
+        self._clear_sms_sent()
+        sms_create_origin = SmsSms.create
+        sms_unlink_origin = SmsSms.unlink
+
+        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': '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 == 'wrong_number_format':
+                        res.update(state='wrong_number_format')
+                    elif error and error == 'unregistered':
+                        res.update(state='unregistered')
+                    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':
+                        self._sms.append({
+                            'number': to_send['number'],
+                            'body': to_send['content'],
+                        })
+                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_unlink(records, *args, **kwargs):
+            if sms_allow_unlink:
+                return sms_unlink_origin(records, *args, **kwargs)
+            # hack: instead of unlink, update state to sent for tests
+            else:
+                records.filtered(lambda sms: sms.id in self._new_sms.ids).state = 'sent'
+            return True
+
+        try:
+            with patch.object(SmsApi, '_contact_iap', side_effect=_contact_iap), \
+                    patch.object(SmsSms, 'create', autospec=True, wraps=SmsSms, side_effect=_sms_sms_create), \
+                    patch.object(SmsSms, 'unlink', autospec=True, wraps=SmsSms, side_effect=_sms_sms_unlink):
+                yield
+        finally:
+            pass
+
+    def _clear_sms_sent(self):
+        self._sms = []
+        self._new_sms = self.env['sms.sms'].sudo()
+
+    def _clear_outoing_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. """
+
+    def _find_sms_sent(self, partner, number):
+        if number is None and partner:
+            number = partner.phone_get_sanitized_number()
+        sent_sms = next((sms for sms in self._sms if sms['number'] == number), None)
+        if not sent_sms:
+            raise AssertionError('sent sms not found for %s (number: %s)' % (partner, number))
+        return sent_sms
+
+    def _find_sms_sms(self, partner, number, status):
+        if number is None and partner:
+            number = partner.phone_get_sanitized_number()
+        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 not sms:
+            raise AssertionError('sms.sms not found for %s (number: %s / status %s)' % (partner, number, status))
+        if len(sms) > 1:
+            raise NotImplementedError()
+        return sms
+
+    def assertNoSMS(self):
+        """ Check no sms went through gateway during mock. """
+        self.assertTrue(len(self._new_sms) == 0)
+
+    def assertSMSIapSent(self, numbers, content=None):
+        """ Check sent SMS. 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 assertSMSSent(self, numbers, content=None):
+        """ Deprecated. Remove in 14.4 """
+        return self.assertSMSIapSent(numbers, content=content)
+
+    def assertSMS(self, partner, number, status, error_code=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 error_code: check error code 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)
+        if error_code:
+            self.assertEqual(sms_sms.error_code, error_code)
+        if content is not None:
+            self.assertIn(content, sms_sms.body)
+        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 == 'sent':
+            self.assertSMSIapSent([sms_sms.number], content=content)
+
+    def assertSMSCanceled(self, partner, number, error_code, 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', error_code=error_code, content=content, fields_values=fields_values)
+
+    def assertSMSFailed(self, partner, number, error_code, 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', error_code=error_code, 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):
+        """ Check content of notifications.
+
+          :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 / sent / exception / canceled (sent by default),
+            'failure_type': optional: sms_number_missing / sms_number_format / sms_credit / sms_server
+            }, { ... }]
+        """
+        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:
+            partner = recipient_info.get('partner', self.env['res.partner'])
+            number = recipient_info.get('number')
+            state = recipient_info.get('state', 'sent')
+            if number is None and partner:
+                number = partner.phone_get_sanitized_number()
+
+            notif = notifications.filtered(lambda n: n.res_partner_id == partner and n.sms_number == number and n.notification_status == state)
+            self.assertTrue(notif, 'SMS: not found notification for %s (number: %s, state: %s)' % (partner, number, state))
+
+            if state not in ('sent', 'ready', 'canceled'):
+                self.assertEqual(notif.failure_type, recipient_info['failure_type'])
+            if check_sms:
+                if state == 'sent':
+                    if sent_unlink:
+                        self.assertSMSIapSent([number], content=content)
+                    else:
+                        self.assertSMS(partner, number, 'sent', content=content)
+                elif state == 'ready':
+                    self.assertSMS(partner, number, 'outgoing', content=content)
+                elif state == 'exception':
+                    self.assertSMS(partner, number, 'error', error_code=recipient_info['failure_type'], content=content)
+                elif state == 'canceled':
+                    self.assertSMS(partner, number, 'canceled', error_code=recipient_info['failure_type'], content=content)
+                else:
+                    raise NotImplementedError('Not implemented')
+
+        if messages is not None:
+            for message in messages:
+                self.assertEqual(content, tools.html2plaintext(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, MockSMS):
+
+    @classmethod
+    def setUpClass(cls):
+        super(SMSCommon, cls).setUpClass()
+        cls.user_employee.write({'login': 'employee'})
+
+        # update country to belgium in order to test sanitization of numbers
+        cls.user_employee.company_id.write({'country_id': cls.env.ref('base.be').id})
+
+        # 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.'
+        })
diff --git a/addons/sms/tests/test_sms_template.py b/addons/sms/tests/test_sms_template.py
new file mode 100644
index 00000000..9483ecaf
--- /dev/null
+++ b/addons/sms/tests/test_sms_template.py
@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+
+from odoo.tests.common import SavepointCase, 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
+
+
+@tagged('post_install')
+class TestSmsTemplateAccessRights(SavepointCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.user_admin = mail_new_test_user(cls.env, login='user_system', groups='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)
+
+    @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()