794 lines
29 KiB
Python
794 lines
29 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 json
|
|
import logging
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.urls import reverse
|
|
from django.utils.translation import gettext as _
|
|
|
|
import tests.utils
|
|
from passerelle.apps.choosit.models import ChoositSMSGateway
|
|
from passerelle.apps.ovh.models import OVHSMSGateway
|
|
from passerelle.apps.sfr_dmc.models import SfrDmcGateway
|
|
from passerelle.apps.smsfactor.models import SMSFactorSMSGateway
|
|
from passerelle.base.models import AccessRight, ApiUser, Job, ResourceLog
|
|
from passerelle.sms.models import SMSLog, SMSResource, TrackCreditSMSResource
|
|
from passerelle.utils.jsonresponse import APIError
|
|
from tests.test_manager import login
|
|
from tests.utils import FakedResponse
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
klasses = [
|
|
klass
|
|
for klass in SMSResource.__subclasses__() + TrackCreditSMSResource.__subclasses__()
|
|
if not klass._meta.abstract
|
|
]
|
|
|
|
|
|
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 is 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 tests.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
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
|
|
def test_sms_legacy_retry_after_error(app, connector, freezer):
|
|
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'],
|
|
}
|
|
freezer.move_to('2019-01-01 00:00:00')
|
|
result = app.post_json(path, params=payload)
|
|
assert result.json['err'] == 0
|
|
assert Job.objects.exists()
|
|
assert Job.objects.get().status == 'registered'
|
|
assert not Job.objects.get().after_timestamp
|
|
with tests.utils.mock_url(
|
|
url=connector.URL,
|
|
response={'status': 429, 'message': 'Too much requests.\nPlease retry in 3 seconds.'},
|
|
):
|
|
connector.jobs()
|
|
assert Job.objects.get().status == 'registered'
|
|
assert Job.objects.get().after_timestamp
|
|
|
|
with tests.utils.mock_url(url=connector.URL, response={'status': 100, 'credit_left': 22}):
|
|
connector.jobs()
|
|
assert Job.objects.get().status == 'registered'
|
|
|
|
freezer.move_to('2019-01-01 00:00:11')
|
|
with tests.utils.mock_url(url=connector.URL, response={'status': 100, 'creditLeft': 22}):
|
|
connector.jobs()
|
|
assert Job.objects.get().status == 'completed'
|
|
assert SMSLog.objects.exists()
|
|
|
|
|
|
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 'accessright/add' not 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 = None
|
|
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, SMSFactorSMSGateway], 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(connector.__class__, 'send_msg') as send_function:
|
|
send_function.return_value = 1
|
|
app.post_json(path, params=payload)
|
|
connector.jobs()
|
|
assert SMSLog.objects.filter(
|
|
appname=connector.get_connector_slug(), slug=connector.slug, credits=1
|
|
).exists()
|
|
|
|
with mock.patch.object(connector.__class__, 'send_msg') as send_function:
|
|
send_function.return_value = 2
|
|
app.post_json(path, params=payload)
|
|
connector.jobs()
|
|
assert SMSLog.objects.filter(
|
|
appname=connector.get_connector_slug(), slug=connector.slug, credits=2
|
|
).exists()
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [OVHSMSGateway, SMSFactorSMSGateway], indirect=True)
|
|
def test_sms_job_details_credits(admin_user, app, connector, caplog):
|
|
path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
|
|
payload = {'message': 'plop', 'from': '+33699999999', 'to': ['+33688888888']}
|
|
with mock.patch.object(connector.__class__, 'send_msg') as send_function:
|
|
send_function.return_value = 1
|
|
app.post_json(path, params=payload)
|
|
job1_id = Job.objects.get(method_name='send_job', status='registered').id
|
|
connector.jobs()
|
|
|
|
send_function.return_value = 2
|
|
app.post_json(path, params=payload)
|
|
job2_id = Job.objects.get(method_name='send_job', status='registered').id
|
|
connector.jobs()
|
|
|
|
app = login(app)
|
|
resp = app.get('/manage/%s/%s/jobs/%s/' % (connector.get_connector_slug(), connector.slug, job1_id))
|
|
assert len(resp.pyquery('td:contains("credits-spent")')) == 1
|
|
assert len(resp.pyquery('td:contains("1")')) == 1
|
|
|
|
resp = app.get('/manage/%s/%s/jobs/%s/' % (connector.get_connector_slug(), connector.slug, job2_id))
|
|
assert len(resp.pyquery('td:contains("credits-spent")')) == 1
|
|
assert len(resp.pyquery('td:contains("2")')) == 1
|
|
|
|
|
|
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 = None
|
|
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, SMSFactorSMSGateway], 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:
|
|
send_function.return_value = None
|
|
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 tests.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)
|
|
assert 'Send a test message' not in resp.text
|
|
|
|
app = login(app)
|
|
resp = app.get(url)
|
|
assert 'Send a test message' in resp.text
|
|
assert resp.pyquery('a#sms-test-send')[0].attrib['href'] == reverse(
|
|
'sms-test-send', kwargs={'connector': connector.get_connector_slug(), 'slug': connector.slug}
|
|
)
|
|
|
|
resp = resp.click('Send a test message')
|
|
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 = None
|
|
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 tests.utils.mock_url(ovh_url, resp, 200):
|
|
connector.jobs()
|
|
connector.refresh_from_db()
|
|
assert connector.credit_left == 123
|
|
|
|
resp = app.get(manager_url)
|
|
assert '123' in resp.text
|
|
|
|
# credit update when checking status
|
|
resp = {
|
|
'creditsLeft': 456,
|
|
}
|
|
with tests.utils.mock_url(ovh_url, resp, 200):
|
|
connector.check_status()
|
|
assert connector.credit_left == 456
|
|
|
|
resp = {
|
|
'message': 'This credential is not valid',
|
|
'httpCode': '403 Forbidden',
|
|
'errorCode': 'INVALID_CREDENTIAL',
|
|
}
|
|
with tests.utils.mock_url(ovh_url, resp, 403):
|
|
with pytest.raises(APIError, match='Forbidden'):
|
|
connector.check_status()
|
|
|
|
|
|
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 tests.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'
|
|
|
|
|
|
def test_ovh_token_request_error(admin_user, app):
|
|
connector = OVHSMSGateway.objects.create(
|
|
slug='test-ovh',
|
|
title='Test OVH',
|
|
account='sms-test42',
|
|
application_key='wrong',
|
|
application_secret='oups',
|
|
)
|
|
|
|
app = login(app)
|
|
resp = app.get(connector.get_absolute_url())
|
|
ovh_request_token_url = 'https://eu.api.ovh.com/1.0/auth/credential'
|
|
ovh_response = {'message': 'Invalid application key'}
|
|
|
|
with tests.utils.mock_url(ovh_request_token_url, ovh_response, 401):
|
|
resp = resp.click('request access').follow()
|
|
assert 'error requesting token: Invalid application key.' in resp.text
|
|
|
|
ovh_response = 'not-json'
|
|
with tests.utils.mock_url(ovh_request_token_url, ovh_response, 401):
|
|
resp = resp.click('request access').follow()
|
|
assert 'error requesting token: bad JSON response' in resp.text
|
|
|
|
|
|
@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.form['default_country_code'] = '+33'
|
|
resp.form['default_trunk_prefix'] = 'x'
|
|
resp = resp.form.submit()
|
|
assert resp.html.find('div', {'class': 'errornotice'}).p.text == 'There were errors processing your form.'
|
|
assert [x.text.strip() for x in resp.html.find_all('div', {'class': 'error'})] == [
|
|
'The country must only contain numbers',
|
|
'The trunk prefix must only contain numbers',
|
|
'This field is required.',
|
|
]
|
|
resp.form['authorized'] = [SMSResource.FR_METRO, SMSResource.FR_DOMTOM]
|
|
resp.form['default_country_code'] = '33'
|
|
resp.form['default_trunk_prefix'] = '0'
|
|
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 = None
|
|
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 = None
|
|
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'
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
|
|
def test_api_statistics(app, freezer, connector, admin_user):
|
|
resp = app.get('/api/statistics/')
|
|
url = [x for x in resp.json['data'] if x['id'] == 'sms_count_ovh_ovhsmsgateway'][0]['url']
|
|
|
|
assert app.get(url, status=403)
|
|
|
|
login(app)
|
|
resp = app.get(url)
|
|
assert len(resp.json['data']['series'][0]['data']) == 0
|
|
|
|
freezer.move_to('2021-01-01 12:00')
|
|
for i in range(5):
|
|
SMSLog.objects.create(appname='ovh', slug='ovhsmsgateway')
|
|
|
|
freezer.move_to('2021-02-03 13:00')
|
|
for i in range(3):
|
|
SMSLog.objects.create(appname='ovh', slug='ovhsmsgateway')
|
|
|
|
freezer.move_to('2021-02-06 13:00')
|
|
SMSLog.objects.create(appname='ovh', slug='ovhsmsgateway')
|
|
SMSLog.objects.create(appname='ovh', slug='other')
|
|
|
|
resp = app.get(url + '?time_interval=day')
|
|
assert resp.json['data'] == {
|
|
'x_labels': ['2021-01-01', '2021-02-03', '2021-02-06'],
|
|
'series': [{'label': 'SMS Count', 'data': [5, 3, 1]}],
|
|
}
|
|
|
|
resp = app.get(url + '?start=2021-02-04&end=2021-02-07')
|
|
assert resp.json['data'] == {
|
|
'x_labels': ['2021-02-06'],
|
|
'series': [{'label': 'SMS Count', 'data': [1]}],
|
|
}
|
|
|
|
# invalid time_interval
|
|
resp = app.get(url + '?time_interval=month')
|
|
assert resp.json['err'] == 1
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [SfrDmcGateway], indirect=True)
|
|
def test_sfr_max_message_error(connector):
|
|
with pytest.raises(APIError, match='you can\'t send more than 20 sms at once.'):
|
|
connector.send_msg('hello', '0033699999999', [f'00336888888{suffix:02d}' for suffix in range(0, 21)])
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [SfrDmcGateway], indirect=True)
|
|
def test_sfr_prefix(connector):
|
|
json_response = {
|
|
'success': True,
|
|
'response': {
|
|
0: '4500283767',
|
|
1: '4500283767',
|
|
},
|
|
}
|
|
|
|
with mock.patch('passerelle.utils.Request.get') as mocked_get:
|
|
mocked_get.return_value = FakedResponse(content=json.dumps(json_response), status_code=200)
|
|
connector.send_msg('hello', '0033699999999', ['0033688888888'])
|
|
mocked_get.assert_called_once_with(
|
|
'https://www.dmc.sfr-sh.fr/DmcWS/1.5.7/JsonService/MessagesUnitairesWS/batchSingleCall',
|
|
params={
|
|
'authenticate': json.dumps(
|
|
{'serviceId': '1234', 'servicePassword': 'krascuky', 'spaceId': '1234'}
|
|
),
|
|
'messageUnitaires': json.dumps(
|
|
[
|
|
{
|
|
'media': 'SMSLong',
|
|
'textMsg': 'hello',
|
|
'from': '0033699999999',
|
|
'to': '+33688888888',
|
|
},
|
|
]
|
|
),
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize('connector', [SfrDmcGateway], indirect=True)
|
|
def test_sfr_unicode_message(connector):
|
|
json_response = {
|
|
'success': True,
|
|
'response': {
|
|
0: '4500283767',
|
|
1: '4500283767',
|
|
},
|
|
}
|
|
|
|
def _check_media_type(message, expected_media_type):
|
|
with mock.patch('passerelle.utils.Request.get') as mocked_get:
|
|
mocked_get.return_value = FakedResponse(content=json.dumps(json_response), status_code=200)
|
|
connector.send_msg(message, '0033699999999', ['0033688888888'])
|
|
mocked_get.assert_called_once()
|
|
messages = json.loads(mocked_get.call_args[1]['params']['messageUnitaires'])
|
|
assert messages[0]['media'] == expected_media_type
|
|
|
|
_check_media_type('standard GSM message', 'SMSLong')
|
|
_check_media_type('usual standard GSM characters : \'"-\r\n!?éèù%à*+=€@[]|', 'SMSLong')
|
|
_check_media_type('unicode message 😀', 'SMSUnicodeLong')
|
|
|
|
|
|
credit_klasses = TrackCreditSMSResource.__subclasses__()
|
|
|
|
|
|
@pytest.fixture(params=credit_klasses)
|
|
def track_credit_connector(request, db):
|
|
klass = request.param
|
|
kwargs = getattr(klass, 'TEST_CREDIT_LEFT', {}).get('create_kwargs', {})
|
|
kwargs.update(
|
|
{
|
|
'title': klass.__name__,
|
|
'slug': klass.__name__.lower(),
|
|
'description': klass.__name__,
|
|
'credit_threshold_alert': 100,
|
|
'credit_left': 102,
|
|
'alert_emails': ['test@entrouvert.org'],
|
|
}
|
|
)
|
|
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_send_alert_emails(track_credit_connector, app, freezer, mailoutbox):
|
|
get_credit_left_payload = track_credit_connector.TEST_CREDIT_LEFT['get_credit_left_payload']
|
|
url = track_credit_connector.TEST_CREDIT_LEFT['url']
|
|
|
|
freezer.move_to('2019-01-01 00:00:00')
|
|
resp = get_credit_left_payload(101)
|
|
with tests.utils.mock_url(url, resp, 200):
|
|
track_credit_connector.check_status()
|
|
assert len(mailoutbox) == 0
|
|
|
|
resp = get_credit_left_payload(99)
|
|
with tests.utils.mock_url(url, resp, 200):
|
|
track_credit_connector.check_status()
|
|
assert len(mailoutbox) == 1
|
|
|
|
mail = mailoutbox[0]
|
|
assert mail.recipients() == ['test@entrouvert.org']
|
|
assert mail.subject == 'SMS alert: only 99 credits left'
|
|
for body in (mail.body, mail.alternatives[0][0]):
|
|
assert track_credit_connector.title in body
|
|
assert track_credit_connector.get_absolute_url() in body
|
|
mailoutbox.clear()
|
|
|
|
# alert is sent again daily
|
|
freezer.move_to('2019-01-01 12:00:00')
|
|
resp = get_credit_left_payload(99)
|
|
with tests.utils.mock_url(url, resp, 200):
|
|
track_credit_connector.check_status()
|
|
assert len(mailoutbox) == 0
|
|
|
|
freezer.move_to('2019-01-02 01:00:07')
|
|
with tests.utils.mock_url(url, resp, 200):
|
|
track_credit_connector.check_status()
|
|
assert len(mailoutbox) == 1
|