292 lines
11 KiB
Python
292 lines
11 KiB
Python
import isodate
|
|
import mock
|
|
import pytest
|
|
from requests import RequestException
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.urlresolvers import reverse
|
|
|
|
from passerelle.apps.ovh.models import OVHSMSGateway
|
|
from passerelle.base.models import ApiUser, AccessRight, Job
|
|
from passerelle.sms.models import SMSResource, SMSLog
|
|
from passerelle.utils.jsonresponse import APIError
|
|
|
|
from test_manager import login, admin_user
|
|
|
|
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'])
|
|
|
|
|
|
@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):
|
|
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.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)
|
|
|
|
|
|
@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'):
|
|
with mock.patch.object(connector, 'send_msg') 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)
|
|
|
|
|
|
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):
|
|
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()
|
|
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
|