Rémi Rahir
7287800101
Since January 2020, users are required to validate their IAP account via SMS code validation. This new behaviour needed to be properly ported to the client to correctly inform the user. This commit adds support of a new failure_type: 'Unregistered Account'. Task #2209567 closes odoo/odoo#48512 Signed-off-by: Thibault Delavallee (tde) <tde@openerp.com>
183 lines
8.5 KiB
Python
183 lines
8.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from contextlib import contextmanager
|
|
from unittest.mock import patch
|
|
|
|
from odoo import exceptions, tools
|
|
from odoo.addons.phone_validation.tools import phone_validation
|
|
from odoo.tests import common
|
|
from odoo.addons.sms.models.sms_api import SmsApi
|
|
|
|
|
|
class MockSMS(common.BaseCase):
|
|
|
|
def tearDown(self):
|
|
super(MockSMS, self).tearDown()
|
|
self._clear_sms_sent
|
|
|
|
@contextmanager
|
|
def mockSMSGateway(self, sim_error=None, nbr_t_error=None):
|
|
self._sms = []
|
|
|
|
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
|
|
|
|
try:
|
|
with patch.object(SmsApi, '_contact_iap', side_effect=_contact_iap) as contact_iap_mock:
|
|
yield
|
|
finally:
|
|
pass
|
|
|
|
def _clear_sms_sent(self):
|
|
self._sms = []
|
|
|
|
def assertSMSSent(self, numbers, content):
|
|
""" 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])))
|
|
self.assertEqual(sent_sms['body'], content)
|
|
|
|
def assertSMSCanceled(self, partner, number, error_code, content=None):
|
|
""" Check canceled SMS. Search is done for a pair partner / number where
|
|
partner can be an empty recordset. """
|
|
if number is None and partner:
|
|
number = partner.phone_get_sanitized_number()
|
|
sms = self.env['sms.sms'].sudo().search([
|
|
('partner_id', '=', partner.id), ('number', '=', number),
|
|
('state', '=', 'canceled')
|
|
])
|
|
self.assertTrue(sms, 'SMS: not found canceled SMS for %s (number: %s)' % (partner, number))
|
|
self.assertEqual(sms.error_code, error_code)
|
|
if content is not None:
|
|
self.assertEqual(sms.body, content)
|
|
|
|
def assertSMSFailed(self, partner, number, error_code, content=None):
|
|
""" Check failed SMS. Search is done for a pair partner / number where
|
|
partner can be an empty recordset. """
|
|
if number is None and partner:
|
|
number = partner.phone_get_sanitized_number()
|
|
sms = self.env['sms.sms'].sudo().search([
|
|
('partner_id', '=', partner.id), ('number', '=', number),
|
|
('state', '=', 'error')
|
|
])
|
|
self.assertTrue(sms, 'SMS: not found failed SMS for %s (number: %s)' % (partner, number))
|
|
self.assertEqual(sms.error_code, error_code)
|
|
if content is not None:
|
|
self.assertEqual(sms.body, content)
|
|
|
|
def assertSMSOutgoing(self, partner, number, content=None):
|
|
""" Check outgoing SMS. Search is done for a pair partner / number where
|
|
partner can be an empty recordset. """
|
|
if number is None and partner:
|
|
number = partner.phone_get_sanitized_number()
|
|
sms = self.env['sms.sms'].sudo().search([
|
|
('partner_id', '=', partner.id), ('number', '=', number),
|
|
('state', '=', 'outgoing')
|
|
])
|
|
self.assertTrue(sms, 'SMS: not found failed SMS for %s (number: %s)' % (partner, number))
|
|
if content is not None:
|
|
self.assertEqual(sms.body, content)
|
|
|
|
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):
|
|
""" 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':
|
|
self.assertSMSSent([number], content)
|
|
elif state == 'ready':
|
|
self.assertSMSOutgoing(partner, number, content)
|
|
elif state == 'exception':
|
|
self.assertSMSFailed(partner, number, recipient_info['failure_type'], content)
|
|
elif state == 'canceled':
|
|
self.assertSMSCanceled(partner, number, recipient_info.get('failure_type', False), 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)
|