entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 15 Releases Activity

orange: update contact everyone SMS connector to API v1.2 (#41092) 

adapted from https://github.com/departement-loire-atlantique/passerelle-orangesms
This commit is contained in:
Nicolas Roche 2020-03-30 14:38:42 +02:00
parent d77739cbec
commit f618ebc937
5 changed files with 323 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
passerelle/apps/orange/migrations/0008_auto_20200412_1240.py Normal file
View File

@ -0,0 +1,47 @@
# -*- coding: utf-8 -*-
# Generated by Django 1.11.18 on 2020-04-12 10:40
from __future__ import unicode_literals
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('orange', '0007_auto_20200310_1539'),
]
operations = [
migrations.RemoveField(
model_name='orangesmsgateway',
name='keystore',
),
migrations.AddField(
model_name='orangesmsgateway',
name='default_country_code',
field=models.CharField(default='33', max_length=3, verbose_name='Préfixe pays'),
),
migrations.AddField(
model_name='orangesmsgateway',
name='default_trunk_prefix',
field=models.CharField(default='0', max_length=2, verbose_name='Préfixe supprimé par défaut'),
),
migrations.AddField(
model_name='orangesmsgateway',
name='groupname',
field=models.CharField(default=None, max_length=64, verbose_name='Groupe'),
preserve_default=False,
),
migrations.AddField(
model_name='orangesmsgateway',
name='password',
field=models.CharField(default=None, max_length=64, verbose_name='Mot de passe'),
preserve_default=False,
),
migrations.AddField(
model_name='orangesmsgateway',
name='username',
field=models.CharField(default=None, max_length=64, verbose_name='Identifiant'),
preserve_default=False,
),
]

116
passerelle/apps/orange/models.py
View File

@ -1,30 +1,116 @@
from django.core.files import File
from django.utils.translation import ugettext_lazy as _
# -*- coding: utf-8 -*-
# passerelle - uniform access to multiple data sources and services
#
# MIT License
# Copyright (c) 2020 departement-loire-atlantique
#
# GNU Affero General Public License
# 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/>.
from django.db import models
from django.utils.translation import ugettext_lazy as _
from passerelle.base.models import SMSResource
from passerelle.utils.jsonresponse import APIError
from . import soap
BASE_API = 'https://contact-everyone.orange-business.com/api/v1.2/'
URL_TOKEN = BASE_API + 'oauth/token'
URL_GROUPS = BASE_API + 'groups'
URL_DIFFUSION = BASE_API + 'groups/%s/diffusion-requests'
class OrangeError(APIError):
pass
def get_json(response):
try:
return response.json()
except ValueError:
raise OrangeError('Orange returned Invalid JSON content: %s' % response.text)
class OrangeSMSGateway(SMSResource):
keystore = models.FileField(upload_to='orange', blank=True, null=True,
verbose_name=_('Keystore'),
help_text=_('Certificate and private key in PEM format'))
default_country_code = '33'
username = models.CharField(verbose_name=_('Identifiant'), max_length=64)
password = models.CharField(verbose_name=_('Mot de passe'), max_length=64)
groupname = models.CharField(verbose_name=_('Groupe'), max_length=64)
URL = ('https://www.api-contact-everyone.fr.orange-business.com/ContactEveryone/services'
'/MultiDiffusionWS')
default_country_code = models.CharField(
verbose_name='Préfixe pays', max_length=3, default='33')
default_trunk_prefix = models.CharField(
verbose_name='Préfixe supprimé par défaut', max_length=2, default='0')
manager_view_template_name = 'passerelle/manage/messages_service_view.html'
class Meta:
verbose_name = 'Orange'
verbose_name = _('Orange')
db_table = 'sms_orange'
def get_access_token(self):
headers = {'content-type': 'application/x-www-form-urlencoded'}
params = {'username': self.username, 'password': self.password}
response = self.requests.post(URL_TOKEN, data=params, headers=headers)
if response.status_code != 200:
raise APIError('Bad username or password: %s, %s' % (
response.status_code, response.text))
response_json = get_json(response)
if 'access_token' not in response_json:
raise OrangeError('Orange do not return access token')
return response_json['access_token']
def group_id_from_name(self, access_token):
headers = {'authorization': 'Bearer %s' % access_token}
response = self.requests.get(URL_GROUPS, headers=headers)
if response.status_code != 200:
raise APIError('Bad token: %s, %s' % (
response.status_code, response.text))
response_json = get_json(response)
group_id = None
for group in response_json:
if group['name'] == self.groupname:
group_id = group['id']
break
if not group_id:
raise APIError('Group name not found: ' + self.groupname)
return group_id
def diffusion(self, access_token, group_id, destinations, message):
headers = {
'content-type': 'application/json',
'authorization': 'Bearer %s' % access_token,
}
payload = {
'name': 'Send a SMS from passerelle',
'msisdns': destinations,
'smsParam': {
'encoding': 'GSM7',
'body': message
}
}
response = self.requests.post(
URL_DIFFUSION % group_id, json=payload, headers=headers)
if response.status_code != 201:
raise OrangeError('Orange fails to send SMS: %s, %s' % (
response.status_code, response.text))
return get_json(response)
def send_msg(self, text, sender, destinations, **kwargs):
"""Send a SMS using the Orange provider"""
# unfortunately it lacks a batch API...
destinations = self.clean_numbers(destinations, self.default_country_code)
return soap.ContactEveryoneSoap(instance=self).send_advanced_message(destinations, sender,
text)
'''Send a SMS using the Orange provider'''
destinations = self.clean_numbers(
destinations, self.default_country_code, self.default_trunk_prefix)
access_token = self.get_access_token()
group_id = self.group_id_from_name(access_token)
response = self.diffusion(access_token, group_id, destinations, text)
return response

15
passerelle/apps/orange/orange.pem
View File

@ -1,15 +0,0 @@
-----BEGIN CERTIFICATE-----
MIICWzCCAcQCAQAwDQYJKoZIhvcNAQEEBQAwdjELMAkGA1UEBhMCRlIxDzANBgNV
BAgTBkZyYW5jZTENMAsGA1UEBxMEQ2FlbjEXMBUGA1UEChMORnJhbmNlIFRlbGVj
b20xDDAKBgNVBAsTA0RQUzEMMAoGA1UEAxMDRERQMRIwEAYJKoZIhvcNAQkBFgNE
SU0wHhcNMDYxMTA4MjAyNDUxWhcNMTYxMTA1MjAyNDUxWjB2MQswCQYDVQQGEwJG
UjEPMA0GA1UECBMGRnJhbmNlMQ0wCwYDVQQHEwRDYWVuMRcwFQYDVQQKEw5GcmFu
Y2UgVGVsZWNvbTEMMAoGA1UECxMDRFBTMQwwCgYDVQQDEwNERFAxEjAQBgkqhkiG
9w0BCQEWA0RJTTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAxp5nnfYr4ZsV
QgIzYkz42Urc+0z49cm8JL5DQAcceIUpWYnCOSjDCEivvkYlBGEaSQhx6goLgpAk
6264BhFIa9tFJBz0VCbZ5erGANNFpi1zK9nglfGMkfgQmPXFcVF+hi9ztff+WHGR
SknuxzXAICG0/PfPy/LcpVC9E35IkG8CAwEAATANBgkqhkiG9w0BAQQFAAOBgQAk
plMn6da0Yu2YZ7dSP9UBrWygN3iD93Krk5H9KcJCFKXRcZsKw/871J+fFxOFxe5u
l/wraMBF+oo9aMBIsrHwzPkPr6/T3+cYScJAcoP0vRqGjbhio1BvoSvH4lsfmJsF
L9cgc58xgDNwztKHqggDtiFCWVEBpYk2jbMnoy7/xg==
-----END CERTIFICATE-----

82
passerelle/apps/orange/soap.py
View File

@ -1,82 +0,0 @@
import datetime
import os.path
from passerelle.utils.jsonresponse import APIError
from passerelle.soap import Soap
from passerelle.xml_builder import XmlBuilder
class ContactEveryoneSoap(Soap):
WSDL_URL = ('https://www.api-contact-everyone.fr.orange-business.com/'
'ContactEveryone/services/MultiDiffusionWS?wsdl')
ORANGE_CERTIFICATE = os.path.join(os.path.dirname(__file__), 'orange.pem')
url = WSDL_URL
class ProfileListBuilder(XmlBuilder):
schema = (
'PROFILE_LIST',
('?loop', 'recipients',
('PROFILE',
('DEST_NAME', 'name_{to}'),
('DEST_FORENAME', 'forename_{to}'),
('DEST_ID', 'ID_{to}'),
('TERMINAL_GROUP',
('TERMINAL',
('TERMINAL_NAME', 'mobile1'),
('TERMINAL_ADDR', '{to}'),
('MEDIA_TYPE_GROUP',
('MEDIA_TYPE', 'sms')))))))
encoding = 'latin1'
@property
def verify(self):
# Do not break if certificate is not updated
if datetime.datetime.now() < datetime.datetime(2016, 11, 5):
return self.ORANGE_CERTIFICATE
else:
return False
def send_message(self, recipients, content):
try:
client = self.get_client()
except Exception as e:
raise APIError('Orange error: WSDL retrieval failed, %s' % e)
message = client.factory.create('WSMessage')
message.fullContenu = True
message.content = content
message.subject = content
message.resumeContent = content
message.strategy = 'sms'
send_profiles = self.ProfileListBuilder().string(
context={'recipients': [{'to': to} for to in recipients]})
message.sendProfiles = send_profiles
try:
resp = client.service.sendMessage(message)
except Exception as e:
raise APIError('Orange error: %s' % e)
else:
return {'msg_ids': [msg_id for msg_id in resp.msgId]}
def send_advanced_message(self, recipients, sender, content):
try:
client = self.get_client()
except Exception as e:
raise APIError('Orange error: WSDL retrieval failed, %s' % e)
message = client.factory.create('WSAdvancedMessage')
message.fullContenu = True
message.content = content
message.subject = content
message.resumeContent = content
message.strategy = 'sms'
message.smsReplyTo = sender
send_profiles = self.ProfileListBuilder().string(
context={'recipients': [{'to': to} for to in recipients]})
message.sendProfiles = send_profiles
try:
resp = client.service.sendAdvancedMessage(message)
except Exception as e:
raise APIError('Orange error: %s' % e)
else:
return {'msg_ids': [msg_id for msg_id in resp.msgId]}

175
tests/test_orange.py Normal file
View File

@ -0,0 +1,175 @@
# -*- coding: utf-8 -*-
# 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_text
from passerelle.apps.orange.models import OrangeSMSGateway, OrangeError
from passerelle.base.models import ApiUser, AccessRight
from passerelle.utils.jsonresponse import APIError
NETLOC = 'contact-everyone.orange-business.com'
JSON_HEADERS = {'content-type': 'application/json'}
PAYLOAD = {
'message': 'hello',
'from': '+33699999999',
'to': ['+33688888888', '+33677777777'],
}
@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_text(request.body))
assert request_body['smsParam']['body'] == PAYLOAD['message']
'33688888888' in request_body['msisdns'][0]
'33677777777' 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'])
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'])
# 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'])
def test_send_msg(app, connector):
url = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
with httmock.HTTMock(response_token_ok, response_group_ok, response_diffusion_ok):
resp = app.post_json(url, params=PAYLOAD, status=200)
assert not resp.json['err']
assert resp.json['data']['status'] == "I'm ok"
# not 201
with httmock.HTTMock(response_token_ok, response_group_ok, response_500):
resp = app.post_json(url, params=PAYLOAD, status=200)
assert resp.json['err']
assert resp.json['err_desc'] == 'Orange fails to send SMS: 500, my_error'