# 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 . 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']