571 lines
21 KiB
Python
571 lines
21 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# Copyright (C) 2020 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
import isodate
|
|
import json
|
|
import logging
|
|
import mock
|
|
import pytest
|
|
from requests import RequestException
|
|
|
|
from django.conf import settings
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.urls import reverse
|
|
from django.utils.translation import ugettext as _
|
|
|
|
from passerelle.apps.choosit.models import ChoositSMSGateway
|
|
from passerelle.apps.ovh.models import OVHSMSGateway
|
|
from passerelle.base.models import ApiUser, AccessRight, Job, ResourceLog
|
|
from passerelle.sms.models import SMSResource, SMSLog
|
|
from passerelle.utils.jsonresponse import APIError
|
|
|
|
from test_manager import login
|
|
|
|
import utils
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
klasses = SMSResource.__subclasses__()
|
|
|
|
|
|
def test_clean_numbers():
|
|
connector = OVHSMSGateway()
|
|
assert connector.clean_numbers(['+ 33 12']) == ['003312']
|
|
assert connector.clean_numbers(['0 0 33 12']) == ['003312']
|
|
assert connector.clean_numbers(['0 12']) == ['003312']
|
|
connector.default_country_code = '32'
|
|
connector.default_trunk_prefix = '1'
|
|
connector.save()
|
|
assert connector.clean_numbers(['+ 33 12']) == ['003312']
|
|
assert connector.clean_numbers(['0 0 33 12']) == ['003312']
|
|
assert connector.clean_numbers(['1 12']) == ['003212']
|
|
with pytest.raises(APIError, match='phone number %r is unsupported' % '0123'):
|
|
connector.clean_numbers(['0123'])
|
|
|
|
|
|
def test_authorize_numbers():
|
|
connector = OVHSMSGateway()
|
|
|
|
# premium-rate
|
|
assert connector.allow_premium_rate == False
|
|
number = '0033' + '8' + '12345678'
|
|
with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
|
|
connector.authorize_numbers([number])
|
|
connector.allow_premium_rate = True
|
|
connector.save()
|
|
assert connector.authorize_numbers([number])[0] == [number]
|
|
|
|
# All country
|
|
assert connector.authorized == [SMSResource.ALL]
|
|
number = '0033' + '1' + '12345678'
|
|
assert connector.authorize_numbers([number])[0] == [number]
|
|
connector.authorized = [SMSResource.FR_METRO]
|
|
connector.save()
|
|
with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
|
|
connector.authorize_numbers([number])
|
|
|
|
# France
|
|
number = '0033' + '6' + '12345678'
|
|
assert connector.authorize_numbers([number])[0] == [number]
|
|
connector.authorized = [SMSResource.FR_DOMTOM]
|
|
connector.save()
|
|
with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
|
|
connector.authorize_numbers([number])
|
|
|
|
# Dom-Tom
|
|
number = '596596' + '123456'
|
|
assert connector.authorize_numbers([number])[0] == [number]
|
|
connector.authorized = [SMSResource.BE_]
|
|
connector.save()
|
|
with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
|
|
connector.authorize_numbers([number])
|
|
|
|
# Belgian
|
|
number = '0032' + '45' + '1234567'
|
|
assert connector.authorize_numbers([number])[0] == [number]
|
|
connector.authorized = [SMSResource.FR_METRO]
|
|
connector.save()
|
|
with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
|
|
connector.authorize_numbers([number])
|
|
|
|
# Don't raise if authorized destinations are not empty
|
|
connector.allow_premium_rate = False
|
|
connector.authorized = [SMSResource.FR_METRO]
|
|
connector.save()
|
|
numbers = [
|
|
'0033' + '8' + '12345678',
|
|
'0033' + '1' + '12345678',
|
|
'0033' + '6' + '12345678',
|
|
'596596' + '123456',
|
|
'0032' + '45' + '1234567',
|
|
]
|
|
authorized_numbers, warnings = connector.authorize_numbers(numbers)
|
|
assert authorized_numbers == ['0033612345678']
|
|
assert warnings == {
|
|
'deny premium rate phone numbers': '0033812345678',
|
|
'deny foreign phone numbers': '0032451234567, 0033112345678, 596596123456',
|
|
}
|
|
|
|
|
|
@pytest.fixture(params=klasses)
|
|
def connector(request, db):
|
|
klass = request.param
|
|
kwargs = getattr(klass, 'TEST_DEFAULTS', {}).get('create_kwargs', {})
|
|
kwargs.update(
|
|
{
|
|
'title': klass.__name__,
|
|
'slug': klass.__name__.lower(),
|
|
'description': klass.__name__,
|
|
}
|
|
)
|
|
c = klass.objects.create(**kwargs)
|
|
api = ApiUser.objects.create(username='apiuser', fullname='Api User', description='api')
|
|
obj_type = ContentType.objects.get_for_model(c)
|
|
# no access check
|
|
AccessRight.objects.create(
|
|
codename='can_send_messages', apiuser=api, resource_type=obj_type, resource_pk=c.pk
|
|
)
|
|
return c
|
|
|
|
|
|
def test_connectors(app, connector, freezer):
|
|
path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
|
|
result = app.post_json(path, params={}, status=400)
|
|
assert result.json['err'] == 1
|
|
assert result.json['err_desc'] == "'message' is a required property"
|
|
|
|
payload = {
|
|
'message': 'hello',
|
|
'from': '+33699999999',
|
|
'to': ['+33688888888', '+33677777777'],
|
|
}
|
|
test_vectors = getattr(connector, 'TEST_DEFAULTS', {}).get('test_vectors', [])
|
|
total = len(test_vectors)
|
|
nb_failed = 0
|
|
assert Job.objects.filter(method_name='send_job').count() == 0
|
|
for test_vector in test_vectors:
|
|
|
|
# register job
|
|
freezer.move_to('2019-01-01 00:00:00')
|
|
result = app.post_json(path, params=payload)
|
|
assert result.json['err'] == 0
|
|
job_id = Job.objects.get(method_name='send_job', status='registered').id
|
|
|
|
# perform job
|
|
freezer.move_to('2019-01-01 01:00:03')
|
|
with utils.mock_url(
|
|
connector.URL, test_vector.get('response', ''), test_vector.get('status_code', 200)
|
|
):
|
|
connector.jobs()
|
|
job = Job.objects.get(id=job_id)
|
|
if job.status == 'failed':
|
|
assert len(job.status_details['error_summary']) > 0
|
|
assert test_vector['result']['err_desc'] in job.status_details['error_summary']
|
|
nb_failed += 1
|
|
else:
|
|
assert job.status == 'completed'
|
|
assert Job.objects.filter(method_name='send_job').count() == total
|
|
assert SMSLog.objects.count() == total - nb_failed
|
|
|
|
|
|
def test_manage_views(admin_user, app, connector):
|
|
url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
|
|
resp = app.get(url)
|
|
assert 'Endpoints' in resp.text
|
|
assert not 'accessright/add' in resp.text
|
|
app = login(app)
|
|
resp = app.get(url)
|
|
description_fields = [
|
|
x.text.split(':')[0] for x in resp.html.find('div', {'id': 'description'}).find_all('p')
|
|
]
|
|
assert 'Default country code' in description_fields
|
|
assert 'Default trunk prefix' in description_fields
|
|
assert 'Maximum message length' in description_fields
|
|
assert 'Account' not in description_fields
|
|
assert 'Username' not in description_fields
|
|
assert 'Endpoints' in resp.text
|
|
assert 'accessright/add' in resp.text
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
|
|
def test_manage_views_ovh(app, connector, admin_user):
|
|
login(app)
|
|
connector.default_country_code = '44'
|
|
connector.account = 'secret'
|
|
connector.application_key = 'secret'
|
|
connector.application_secret = 'secret'
|
|
connector.consumer_key = 'secret'
|
|
connector.password = 'secret'
|
|
connector.username = 'secret'
|
|
connector.alert_emails = ['test@entrouvert.org', 'foo@example.com']
|
|
connector.save()
|
|
url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
|
|
resp = app.get(url)
|
|
description_fields = [x.text for x in resp.html.find('div', {'id': 'description'}).find_all('p')]
|
|
assert any(x for x in description_fields if 'Default country code' in x)
|
|
assert any(x for x in description_fields if '44' in x)
|
|
assert not any(x for x in description_fields if 'secret' in x)
|
|
alert_emails_filed = [x for x in description_fields if 'send credit alerts to' in x]
|
|
assert alert_emails_filed[0].split(':')[1].strip() == 'test@entrouvert.org, foo@example.com'
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
|
|
def test_sms_max_message_length(app, connector):
|
|
path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
|
|
|
|
message_above_limit = 'a' * (connector.max_message_length + 1)
|
|
payload = {
|
|
'message': message_above_limit,
|
|
'from': '+33699999999',
|
|
'to': ['+33688888888'],
|
|
}
|
|
with mock.patch.object(OVHSMSGateway, 'send_msg') as send_function:
|
|
send_function.return_value = {}
|
|
result = app.post_json(path, params=payload)
|
|
connector.jobs()
|
|
assert send_function.call_args[1]['text'] == 'a' * connector.max_message_length
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
|
|
def test_sms_log(app, connector):
|
|
path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
|
|
assert not SMSLog.objects.filter(appname=connector.get_connector_slug(), slug=connector.slug).exists()
|
|
|
|
payload = {
|
|
'message': 'plop',
|
|
'from': '+33699999999',
|
|
'to': ['+33688888888'],
|
|
}
|
|
with mock.patch.object(OVHSMSGateway, 'send_msg') as send_function:
|
|
send_function.return_value = {}
|
|
result = app.post_json(path, params=payload)
|
|
connector.jobs()
|
|
assert SMSLog.objects.filter(appname=connector.get_connector_slug(), slug=connector.slug).exists()
|
|
|
|
|
|
def test_sms_nostop_parameter(app, connector):
|
|
base_path = '/%s/%s/send/?nostop=1' % (connector.get_connector_slug(), connector.slug)
|
|
payload = {
|
|
'message': 'not a spam',
|
|
'from': '+33699999999',
|
|
'to': ['+33688888888'],
|
|
}
|
|
for path in (base_path, base_path + '?nostop=1', base_path + '?nostop=foo', base_path + '?nostop'):
|
|
send_patch = mock.patch(
|
|
'passerelle.apps.%s.models.%s.send_msg'
|
|
% (connector.__class__._meta.app_label, connector.__class__.__name__)
|
|
)
|
|
with send_patch as send_function:
|
|
send_function.return_value = {}
|
|
result = app.post_json(base_path, params=payload)
|
|
connector.jobs()
|
|
assert send_function.call_args[1]['text'] == 'not a spam'
|
|
assert send_function.call_args[1]['stop'] == ('nostop' not in path)
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
|
|
@pytest.mark.parametrize(
|
|
'to, destination',
|
|
[
|
|
('06 12 34 56 78', '0033612345678'),
|
|
('06.12.34.56.78', '0033612345678'),
|
|
('06-12-34-56-78', '0033612345678'),
|
|
('+33/612345678', '0033612345678'),
|
|
],
|
|
)
|
|
def test_send_schema(app, connector, to, destination):
|
|
base_path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
|
|
payload = {
|
|
'message': 'not a spam',
|
|
'from': '+33699999999',
|
|
'to': [to],
|
|
}
|
|
send_patch = mock.patch(
|
|
'passerelle.apps.%s.models.%s.send_msg'
|
|
% (connector.__class__._meta.app_label, connector.__class__.__name__)
|
|
)
|
|
with send_patch as send_function:
|
|
app.post_json(base_path, params=payload)
|
|
connector.jobs()
|
|
assert send_function.call_args[1]['destinations'] == [destination]
|
|
|
|
|
|
def test_ovh_new_api(app, freezer):
|
|
connector = OVHSMSGateway.objects.create(
|
|
slug='ovh',
|
|
account='sms-test42',
|
|
application_key='RHrTdU2oTsrVC0pu',
|
|
application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
|
|
consumer_key='iF0zi0MJrbjNcI3hvuvwkhNk8skrigxz',
|
|
)
|
|
api = ApiUser.objects.create(username='apiuser')
|
|
obj_type = ContentType.objects.get_for_model(connector)
|
|
# no access check
|
|
AccessRight.objects.create(
|
|
codename='can_send_messages', apiuser=api, resource_type=obj_type, resource_pk=connector.pk
|
|
)
|
|
|
|
payload = {
|
|
'message': 'hello',
|
|
'from': '+33699999999',
|
|
'to': ['+33688888888', '+33677777777'],
|
|
}
|
|
|
|
# register job
|
|
freezer.move_to('2019-01-01 00:00:00')
|
|
path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
|
|
result = app.post_json(path, params=payload)
|
|
assert result.json['err'] == 0
|
|
job_id = Job.objects.get(method_name='send_job', status='registered').id
|
|
|
|
# perform job
|
|
freezer.move_to('2019-01-01 01:00:03')
|
|
resp = {
|
|
'validReceivers': ['+33688888888', '+33677777777'],
|
|
'totalCreditsRemoved': 1,
|
|
'ids': [241615100],
|
|
'invalidReceivers': [],
|
|
}
|
|
base_url = connector.API_URL % {'serviceName': 'sms-test42'}
|
|
url = base_url + 'jobs/'
|
|
with utils.mock_url(url, resp, 200) as mocked:
|
|
connector.jobs()
|
|
job = Job.objects.get(id=job_id)
|
|
assert job.status == 'completed'
|
|
|
|
request = mocked.handlers[0].call['requests'][0]
|
|
assert 'X-Ovh-Signature' in request.headers
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
|
|
def test_sms_test_send(admin_user, app, connector):
|
|
url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
|
|
resp = app.get(url)
|
|
link = resp.html.find('div', {'id': 'endpoints'}).find_all('a')[-1]
|
|
assert 'Send a test message' not in link.text
|
|
|
|
app = login(app)
|
|
resp = app.get(url)
|
|
link = resp.html.find('div', {'id': 'endpoints'}).find_all('a')[-1]
|
|
assert 'Send a test message' in link.text
|
|
assert link['href'] == reverse(
|
|
'sms-test-send', kwargs={'connector': connector.get_connector_slug(), 'slug': connector.slug}
|
|
)
|
|
|
|
resp = app.get(link['href'])
|
|
resp.form['number'] = '+33688888888'
|
|
resp.form['sender'] = '+33699999999'
|
|
resp.form['message'] = 'hello'
|
|
with mock.patch.object(OVHSMSGateway, 'send_msg') as send_function:
|
|
send_function.return_value = {}
|
|
resp = resp.form.submit()
|
|
assert send_function.call_args[1] == {
|
|
'text': 'hello',
|
|
'sender': '+33699999999',
|
|
'destinations': ['0033688888888'],
|
|
'stop': False,
|
|
}
|
|
assert resp.status_code == 302
|
|
assert resp.location == url
|
|
|
|
|
|
def test_ovh_new_api_credit(app, freezer, admin_user):
|
|
login(app)
|
|
connector = OVHSMSGateway.objects.create(
|
|
slug='ovh',
|
|
account='sms-test42',
|
|
application_key='RHrTdU2oTsrVC0pu',
|
|
application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
|
|
consumer_key='iF0zi0MJrbjNcI3hvuvwkhNk8skrigxz',
|
|
)
|
|
|
|
manager_url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
|
|
resp = app.get(manager_url)
|
|
assert 'no credit left' in resp.text
|
|
|
|
# a job to update credit was added on connector creation
|
|
resp = {
|
|
'creditsLeft': 123,
|
|
}
|
|
ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}
|
|
with utils.mock_url(ovh_url, resp, 200) as mocked:
|
|
connector.jobs()
|
|
connector.refresh_from_db()
|
|
assert connector.credit_left == 123
|
|
|
|
resp = app.get(manager_url)
|
|
assert '123' in resp.text
|
|
|
|
# hourly update
|
|
resp = {
|
|
'creditsLeft': 456,
|
|
}
|
|
with utils.mock_url(ovh_url, resp, 200) as mocked:
|
|
connector.hourly()
|
|
assert connector.credit_left == 456
|
|
|
|
|
|
def test_ovh_alert_emails(app, freezer, mailoutbox):
|
|
connector = OVHSMSGateway.objects.create(
|
|
slug='test-ovh',
|
|
title='Test OVH',
|
|
account='sms-test42',
|
|
application_key='RHrTdU2oTsrVC0pu',
|
|
application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
|
|
consumer_key='iF0zi0MJrbjNcI3hvuvwkhNk8skrigxz',
|
|
credit_threshold_alert=100,
|
|
credit_left=102,
|
|
alert_emails=['test@entrouvert.org'],
|
|
)
|
|
api = ApiUser.objects.create(username='apiuser')
|
|
obj_type = ContentType.objects.get_for_model(connector)
|
|
AccessRight.objects.create(
|
|
codename='can_send_messages', apiuser=api, resource_type=obj_type, resource_pk=connector.pk
|
|
)
|
|
|
|
freezer.move_to('2019-01-01 00:00:00')
|
|
resp = {'creditsLeft': 101}
|
|
ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}
|
|
with utils.mock_url(ovh_url, resp, 200) as mocked:
|
|
connector.hourly()
|
|
assert len(mailoutbox) == 0
|
|
|
|
resp = {'creditsLeft': 99}
|
|
ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}
|
|
with utils.mock_url(ovh_url, resp, 200) as mocked:
|
|
connector.hourly()
|
|
assert len(mailoutbox) == 1
|
|
|
|
mail = mailoutbox[0]
|
|
assert mail.recipients() == ['test@entrouvert.org']
|
|
assert mail.subject == 'OVH SMS alert: only 99 credits left'
|
|
for body in (mail.body, mail.alternatives[0][0]):
|
|
assert connector.account in body
|
|
assert connector.title in body
|
|
assert 'http://localhost/ovh/test-ovh/' in body
|
|
mailoutbox.clear()
|
|
|
|
# alert is sent again daily
|
|
freezer.move_to('2019-01-01 12:00:00')
|
|
resp = {'creditsLeft': 99}
|
|
ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}
|
|
with utils.mock_url(ovh_url, resp, 200) as mocked:
|
|
connector.hourly()
|
|
assert len(mailoutbox) == 0
|
|
|
|
freezer.move_to('2019-01-02 01:00:07')
|
|
with utils.mock_url(ovh_url, resp, 200) as mocked:
|
|
connector.hourly()
|
|
assert len(mailoutbox) == 1
|
|
|
|
|
|
def test_ovh_token_request(admin_user, app):
|
|
connector = OVHSMSGateway.objects.create(
|
|
slug='test-ovh',
|
|
title='Test OVH',
|
|
account='sms-test42',
|
|
application_key='RHrTdU2oTsrVC0pu',
|
|
application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
|
|
)
|
|
|
|
app = login(app)
|
|
resp = app.get(connector.get_absolute_url())
|
|
assert 'not operational yet' in resp.text
|
|
|
|
ovh_request_token_url = 'https://eu.api.ovh.com/1.0/auth/credential'
|
|
ovh_response = {
|
|
'consumerKey': 'xyz',
|
|
'validationUrl': 'https://eu.api.ovh.com/auth/?credentialToken=iQ1joJE',
|
|
}
|
|
with utils.mock_url(ovh_request_token_url, ovh_response, 302) as mocked:
|
|
resp = resp.click('request access')
|
|
assert resp.url == 'https://eu.api.ovh.com/auth/?credentialToken=iQ1joJE'
|
|
|
|
request = mocked.handlers[0].call['requests'][0]
|
|
body = json.loads(request.body.decode())
|
|
assert 'accessRules' in body
|
|
redirect_url = body['redirection'][len('http://testserver') :]
|
|
|
|
resp = app.get(redirect_url).follow()
|
|
assert 'Successfuly completed connector configuration' in resp.text
|
|
connector.refresh_from_db()
|
|
assert connector.consumer_key == 'xyz'
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [ChoositSMSGateway], indirect=True)
|
|
def test_manager(admin_user, app, connector):
|
|
app = login(app)
|
|
path = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
|
|
resp = app.get(path)
|
|
assert (
|
|
'33'
|
|
in [
|
|
x.text
|
|
for x in resp.html.find('div', {'id': 'description'}).find_all('p')
|
|
if x.text.startswith(_('Default country code'))
|
|
][0]
|
|
)
|
|
assert (
|
|
_('All')
|
|
in [x.text for x in resp.html.find_all('p') if x.text.startswith(_('Authorized Countries'))][0]
|
|
)
|
|
assert (
|
|
_('no')
|
|
in [x.text for x in resp.html.find_all('p') if x.text.startswith(_('Allow premium rate numbers'))][0]
|
|
)
|
|
|
|
path = '/manage/%s/%s/edit' % (connector.get_connector_slug(), connector.slug)
|
|
resp = app.get(path)
|
|
resp.form['authorized'] = []
|
|
resp = resp.form.submit()
|
|
assert resp.html.find('div', {'class': 'errornotice'}).p.text == 'There were errors processing your form.'
|
|
assert resp.html.find('div', {'class': 'error'}).text.strip() == 'This field is required.'
|
|
resp.form['authorized'] = [SMSResource.FR_METRO, SMSResource.FR_DOMTOM]
|
|
resp = resp.form.submit()
|
|
resp = resp.follow()
|
|
assert (
|
|
_('France mainland (+33 [67])')
|
|
in [x.text for x in resp.html.find_all('p') if x.text.startswith(_('Authorized Countries'))][0]
|
|
)
|
|
|
|
path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
|
|
payload = {
|
|
'message': 'plop',
|
|
'from': '+33699999999',
|
|
'to': ['+33688888888'],
|
|
}
|
|
resp = app.post_json(path, params=payload)
|
|
assert resp.json['warn'] == {
|
|
'deny premium rate phone numbers': '',
|
|
'deny foreign phone numbers': '',
|
|
}
|
|
with mock.patch.object(type(connector), 'send_msg') as send_function:
|
|
send_function.return_value = {}
|
|
connector.jobs()
|
|
assert SMSLog.objects.count() == 1
|
|
|
|
payload['to'][0] = '+33188888888'
|
|
SMSLog.objects.all().delete()
|
|
app.post_json(path, params=payload)
|
|
with mock.patch.object(type(connector), 'send_msg') as send_function:
|
|
send_function.return_value = {}
|
|
connector.jobs()
|
|
assert not SMSLog.objects.count()
|
|
assert ResourceLog.objects.filter(levelno=logging.WARNING).count() == 1
|
|
assert (
|
|
ResourceLog.objects.filter(levelno=30)[0].extra['exception']
|
|
== 'no phone number was authorized: 0033188888888'
|
|
)
|