207 lines
7.4 KiB
Python
207 lines
7.4 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 httmock
|
|
import pytest
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.utils.encoding import force_str
|
|
|
|
from passerelle.apps.orange.models import OrangeError, OrangeSMSGateway
|
|
from passerelle.base.models import AccessRight, ApiUser, Job
|
|
from passerelle.utils.jsonresponse import APIError
|
|
|
|
NETLOC = 'contact-everyone.orange-business.com'
|
|
JSON_HEADERS = {'content-type': 'application/json'}
|
|
PAYLOAD = {
|
|
'message': 'hello',
|
|
'from': 'john',
|
|
'to': ['+33677777777', '+33688888888'],
|
|
}
|
|
|
|
|
|
@httmock.urlmatch(netloc=NETLOC, path='/api/v1.2/oauth/token', method='POST')
|
|
def response_token_ok(url, request):
|
|
assert 'username=jdoe' in request.body
|
|
assert 'password=secret' in request.body
|
|
content = json.dumps({'access_token': 'my_token'})
|
|
return httmock.response(200, content, JSON_HEADERS)
|
|
|
|
|
|
@httmock.urlmatch(netloc=NETLOC, path='/api/v1.2/groups', method='GET')
|
|
def response_group_ok(url, request):
|
|
content = json.dumps(
|
|
[
|
|
{'name': 'group1', 'id': 'gid1'},
|
|
{'name': 'group2', 'id': 'gid2'},
|
|
]
|
|
)
|
|
return httmock.response(200, content, JSON_HEADERS)
|
|
|
|
|
|
@httmock.urlmatch(netloc=NETLOC, path='/api/v1.2/groups/gid2/diffusion-requests', method='POST')
|
|
def response_diffusion_ok(url, request):
|
|
assert request.headers['authorization'] == 'Bearer my_token'
|
|
request_body = json.loads(force_str(request.body))
|
|
assert request_body['smsParam']['body'] == PAYLOAD['message']
|
|
assert 'senderName' not in request_body['smsParam'].keys()
|
|
assert '33677777777' in request_body['msisdns'][0]
|
|
assert '33688888888' in request_body['msisdns'][1]
|
|
content = json.dumps({'status': "I'm ok"})
|
|
return httmock.response(201, content, JSON_HEADERS)
|
|
|
|
|
|
@httmock.urlmatch(netloc=NETLOC)
|
|
def response_500(url, request):
|
|
return httmock.response(500, 'my_error')
|
|
|
|
|
|
@httmock.urlmatch(netloc=NETLOC)
|
|
def response_invalid_json(url, request):
|
|
return httmock.response(200, 'not a JSON content')
|
|
|
|
|
|
def setup_access_rights(obj):
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(obj)
|
|
AccessRight.objects.create(
|
|
codename='can_send_messages', apiuser=api, resource_type=obj_type, resource_pk=obj.pk
|
|
)
|
|
return obj
|
|
|
|
|
|
@pytest.fixture
|
|
def connector(db):
|
|
return setup_access_rights(
|
|
OrangeSMSGateway.objects.create(
|
|
slug='my_connector', username='jdoe', password='secret', groupname='group2'
|
|
)
|
|
)
|
|
|
|
|
|
def test_get_access_token(app, connector):
|
|
orange = OrangeSMSGateway()
|
|
orange.username = 'jdoe'
|
|
orange.password = 'secret'
|
|
with httmock.HTTMock(response_token_ok):
|
|
assert orange.get_access_token() == 'my_token'
|
|
|
|
# not 200
|
|
with pytest.raises(APIError, match='Bad username or password'):
|
|
with httmock.HTTMock(response_500):
|
|
orange.get_access_token()
|
|
|
|
# not json
|
|
with pytest.raises(OrangeError, match='Orange returned Invalid JSON content'):
|
|
with httmock.HTTMock(response_invalid_json):
|
|
orange.get_access_token()
|
|
|
|
# no token
|
|
@httmock.urlmatch(netloc=NETLOC, path='/api/v1.2/oauth/token', method='POST')
|
|
def mocked_response(url, request):
|
|
return httmock.response(200, '{}')
|
|
|
|
with pytest.raises(OrangeError, match='Orange do not return access token'):
|
|
with httmock.HTTMock(mocked_response):
|
|
orange.get_access_token()
|
|
|
|
|
|
def test_group_id_from_name(app, connector):
|
|
orange = OrangeSMSGateway()
|
|
orange.groupname = 'group2'
|
|
with httmock.HTTMock(response_group_ok):
|
|
assert orange.group_id_from_name('my_token') == 'gid2'
|
|
|
|
# no group
|
|
orange.groupname = 'group3'
|
|
with pytest.raises(APIError, match='Group name not found: group3'):
|
|
with httmock.HTTMock(response_group_ok):
|
|
orange.group_id_from_name('my_token')
|
|
|
|
# not 200
|
|
orange.groupname = 'group2'
|
|
with pytest.raises(APIError, match='Bad token'):
|
|
with httmock.HTTMock(response_500):
|
|
orange.group_id_from_name('my_token')
|
|
|
|
# not json
|
|
with pytest.raises(OrangeError, match='Orange returned Invalid JSON content'):
|
|
with httmock.HTTMock(response_invalid_json):
|
|
orange.group_id_from_name('my_token')
|
|
|
|
|
|
def test_diffusion(app, connector):
|
|
orange = OrangeSMSGateway()
|
|
with httmock.HTTMock(response_diffusion_ok):
|
|
resp = orange.diffusion('my_token', 'gid2', PAYLOAD['to'], PAYLOAD['message'], PAYLOAD['from'])
|
|
assert resp['status'] == "I'm ok"
|
|
|
|
# not 201
|
|
with pytest.raises(OrangeError, match='Orange fails to send SMS'):
|
|
with httmock.HTTMock(response_500):
|
|
orange.diffusion('my_token', 'gid2', PAYLOAD['to'], PAYLOAD['message'], PAYLOAD['from'])
|
|
|
|
# not json
|
|
@httmock.urlmatch(netloc=NETLOC)
|
|
def mocked_response(url, request):
|
|
return httmock.response(201, 'not a JSON content')
|
|
|
|
with pytest.raises(OrangeError, match='Orange returned Invalid JSON content'):
|
|
with httmock.HTTMock(mocked_response):
|
|
orange.diffusion('my_token', 'gid2', PAYLOAD['to'], PAYLOAD['message'], PAYLOAD['from'])
|
|
|
|
# sender name not allowed
|
|
@httmock.urlmatch(netloc=NETLOC)
|
|
def mocked_response2(url, request):
|
|
request_body = json.loads(force_str(request.body))
|
|
assert request_body['smsParam']['senderName'] == 'john'
|
|
error_response = [
|
|
{
|
|
'code': 'SenderNameNotAllowed',
|
|
'field': 'smsParam.senderName',
|
|
'message': 'The given sender name is not allowed.',
|
|
}
|
|
]
|
|
return httmock.response(400, json.dumps(error_response))
|
|
|
|
orange.provide_sender = True
|
|
orange.save()
|
|
with pytest.raises(OrangeError, match='The given sender name is not allowed.'):
|
|
with httmock.HTTMock(mocked_response2):
|
|
orange.diffusion('my_token', 'gid2', PAYLOAD['to'], PAYLOAD['message'], PAYLOAD['from'])
|
|
|
|
|
|
def test_send_msg(app, connector):
|
|
url = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
|
|
assert Job.objects.count() == 0
|
|
resp = app.post_json(url, params=PAYLOAD, status=200)
|
|
assert not resp.json['err']
|
|
assert Job.objects.count() == 1
|
|
with httmock.HTTMock(response_token_ok, response_group_ok, response_diffusion_ok):
|
|
connector.jobs()
|
|
assert Job.objects.all()[0].status == 'completed'
|
|
|
|
# not 201
|
|
resp = app.post_json(url, params=PAYLOAD, status=200)
|
|
assert not resp.json['err']
|
|
assert Job.objects.count() == 2
|
|
with httmock.HTTMock(response_token_ok, response_group_ok, response_500):
|
|
connector.jobs()
|
|
job = Job.objects.all()[1]
|
|
assert job.status == 'failed'
|
|
assert 'Orange fails to send SMS: 500, my_error' in job.status_details['error_summary']
|