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

sms: improve error handling of all providers with APIError (fixes #13499) 

All specific exception classes have been removed, sending was fixed in orange,
oxyd and choosit providers; when possible failure for each destination is
returned.
This commit is contained in:
Benjamin Dauvergne 2016-10-07 10:56:48 +02:00
parent fd761d3403
commit 41a5bbaa03
11 changed files with 369 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
passerelle/apps/choosit/models.py
View File

@ -1,26 +1,72 @@
import re
import urllib
import urllib2
import logging
# -*- coding: utf-8 -*-
import json
import requests
from django.utils.translation import ugettext_lazy as _
from django.db import models
from django.core.urlresolvers import reverse
from passerelle.utils.jsonresponse import APIError
from passerelle.base.models import BaseResource
from passerelle.sms import SMSGatewayMixin
from django.core.urlresolvers import reverse
from .choosit import ChoositRegisterWS
class ChoositError(Exception):
pass
class ChoositSMSGateway(BaseResource, SMSGatewayMixin):
key = models.CharField(max_length=64)
default_country_code = models.CharField(max_length=3, default=u'33')
# FIXME: add regexp field, to check destination and from format
TEST_DEFAULTS = {
'create_kwargs': {
'key': '1234',
},
'test_vectors': [
{
'response': '',
'result': {
'err': 1,
'err_desc': 'Choosit error: some destinations failed',
'data': [
[u'33688888888', u'Choosit error: bad JSON response No JSON object '
'could be decoded'],
[u'33677777777', u'Choosit error: bad JSON response No JSON object '
'could be decoded'],
]
}
},
{
'response': {
'error': 'not ok',
},
'result': {
'err': 1,
'err_desc': 'Choosit error: some destinations failed',
'data': [
[u'33688888888', u'Choosit error: not ok'],
[u'33677777777', u'Choosit error: not ok'],
],
}
},
{
'response': {
'result': u'Envoi terminé',
'sms_id': 1234,
},
'result': {
'err': 0,
'data': [
[u'33688888888', {'result': u'Envoi terminé', 'sms_id': 1234}],
[u'33677777777', {'result': u'Envoi terminé', 'sms_id': 1234}],
],
}
}
],
}
URL = 'http://sms.choosit.com/webservice'
class Meta:
verbose_name = 'Choosit'
db_table = 'sms_choosit'
@ -37,27 +83,35 @@ class ChoositSMSGateway(BaseResource, SMSGatewayMixin):
"""Send a SMS using the Choosit provider"""
# from http://sms.choosit.com/documentation_technique.html
# unfortunately it lacks a batch API...
destinations = self.clean_numbers(destinations,
self.default_country_code, prefix='')
destinations = self.clean_numbers(destinations, self.default_country_code, prefix='')
results = []
for dest in destinations:
params = {
'key': self.key,
'recipient': dest,
'content': text[:160],
}
}
data = {'data': json.dumps(params)}
try:
data = urllib.urlencode([('data', json.dumps(params))])
r = urllib2.Request('http://sms.choosit.com/webservice', data)
p = urllib2.urlopen(r, timeout=5)
output = json.load(p)
p.close()
except Exception, e:
logging.exception(u'%s unable to send message %s to %s',
self, text, destinations)
raise ChoositError(e)
if 'error' in output:
raise ChoositError(output['error'].encode('utf-8'))
return { 'message': output.get('result') }
r = requests.post(self.URL, data=data)
except requests.RequestException as e:
results.append('Choosit error: %s' % e)
else:
try:
output = r.json()
except ValueError as e:
results.append('Choosit error: bad JSON response %s' % e)
else:
if not isinstance(output, dict):
results.append('Choosit error: JSON response is not a dict %r' % output)
elif 'error' in output:
results.append(u'Choosit error: %s' % output['error'])
else:
results.append(output)
if any(isinstance(result, basestring) for result in results):
raise APIError('Choosit error: some destinations failed',
data=zip(destinations, results))
return zip(destinations, results)
class ChoositRegisterNewsletter(models.Model):
@ -77,8 +131,7 @@ class ChoositRegisterGateway(BaseResource):
url = models.CharField(max_length=200)
key = models.CharField(max_length=64)
newsletters = models.ManyToManyField(ChoositRegisterNewsletter,
blank=True)
newsletters = models.ManyToManyField(ChoositRegisterNewsletter, blank=True)
class Meta:
verbose_name = 'Choosit Registration'
@ -110,7 +163,7 @@ class ChoositRegisterGateway(BaseResource):
ws = reg.subscriptions(email=user or '')
if 'error' in ws:
raise ChoositError(ws['error'])
raise APIError(ws['error'])
newsletters = ws['newsletter']
subscriptions = ws['subscriptions']
@ -125,10 +178,10 @@ class ChoositRegisterGateway(BaseResource):
ret = []
for id, name in newsletters.items():
newsletter = {
'name': id,
'description': name,
'transports': { 'available': ['mail'] }
}
'name': id,
'description': name,
'transports': {'available': ['mail']}
}
if user:
if id in subscriptions:
newsletter['transports']['defined'] = ['mail']

66
passerelle/apps/mobyt/models.py
View File

@ -1,38 +1,45 @@
import re
import urllib
import urllib2
import logging
import json
from django.utils.translation import ugettext_lazy as _
from django.db import models
from django.core.urlresolvers import reverse
import requests
from passerelle.utils.jsonresponse import APIError
from passerelle.base.models import BaseResource
from passerelle.sms import SMSGatewayMixin
logger = logging.getLogger('passerelle.apps.mobyt')
class MobytError(Exception):
pass
class MobytSMSGateway(BaseResource, SMSGatewayMixin):
URL = 'http://multilevel.mobyt.fr/sms/batch.php'
MESSAGES_QUALITIES = (
('l', _('sms direct')),
('ll', _('sms low-cost')),
('n', _('sms top')),
('l', _('sms direct')),
('ll', _('sms low-cost')),
('n', _('sms top')),
)
username = models.CharField(max_length=64)
password = models.CharField(max_length=64)
quality = models.CharField(max_length=4, choices=MESSAGES_QUALITIES,
default='l', verbose_name=_('message quality'))
quality = models.CharField(max_length=4, choices=MESSAGES_QUALITIES, default='l',
verbose_name=_('message quality'))
default_country_code = models.CharField(max_length=3, default=u'33')
# FIXME: add regexp field, to check destination and from format
TEST_DEFAULTS = {
'create_kwargs': {
'username': 'john',
'password': 'doe',
},
'test_vectors': [
{
'response': '',
'result': {
'err': 1,
'err_desc': 'MobyT error: received \'\' instead of OK',
}
}
],
}
class Meta:
verbose_name = 'Mobyt'
db_table = 'sms_mobyt'
@ -46,21 +53,18 @@ class MobytSMSGateway(BaseResource, SMSGatewayMixin):
# unfortunately it lacks a batch API...
destinations = self.clean_numbers(destinations, self.default_country_code)
rcpt = ','.join(destinations)
params = urllib.urlencode({
params = {
'user': self.username,
'pass': self.password,
'rcpt': rcpt,
'data': text,
'sender': sender,
'qty': self.quality
})
'qty': self.quality,
}
try:
r = urllib2.urlopen(self.URL, params)
except Exception, e:
logger.error('unable to urlopen %s: %s', self.URL, e)
raise MobytError('unable to urlopen %s: %s' % (self.URL, e))
answer = r.read()
r.close()
if answer[:2] != "OK":
raise MobytError('MobyT error: %r' % answer)
return { 'message': 'OK' }
r = requests.post(self.URL, data=params)
except requests.RequestException, e:
raise APIError('MobyT error: POST failed, %s' % e)
if r.content[:2] != "OK":
raise APIError('MobyT error: received %r instead of OK' % r.content)
return None

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

@ -1,29 +1,24 @@
import re
import urllib
import urllib2
import logging
import json
from StringIO import StringIO
from django.core.files import File
from django.utils.translation import ugettext_lazy as _
from django.db import models
from django.core.urlresolvers import reverse
from passerelle.base.models import BaseResource
from passerelle.sms import SMSGatewayMixin
from . import soap
class OrangeError(Exception):
pass
class OrangeSMSGateway(BaseResource, SMSGatewayMixin):
keystore = models.FileField(upload_to='orange',
blank=True, null=True,
verbose_name=_('Keystore'),
help_text=_('Certificate and private key in PEM format'))
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'
URL = ('https://www.api-contact-everyone.fr.orange-business.com/ContactEveryone/services'
'/MultiDiffusionWS')
class Meta:
verbose_name = 'Orange'
db_table = 'sms_orange'
@ -33,11 +28,8 @@ class OrangeSMSGateway(BaseResource, SMSGatewayMixin):
return 'phone'
def send_msg(self, text, sender, destinations):
logger = logging.getLogger('passerelle.apps.orange')
"""Send a SMS using the Orange provider"""
# unfortunately it lacks a batch API...
destinations = self.clean_numbers(destinations, self.default_country_code)
logger.info(u'sending sms with sender %s to %s', sender, u', '.join(destinations))
logger.debug(u'sms content: %s', text)
#soap.ContactEveryoneSoap(instance=self).send_advanced_message(destinations, sender, text)
return soap.ContactEveryoneSoap(instance=self).send_advanced_message(destinations, sender,
text)

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

@ -1,8 +1,8 @@
import logging
import datetime
import os.path
from passerelle.utils.jsonresponse import APIError
from passerelle.soap import Soap
from passerelle.xml_builder import XmlBuilder
@ -39,7 +39,10 @@ class ContactEveryoneSoap(Soap):
return False
def send_message(self, recipients, content):
client = self.get_client()
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
@ -49,15 +52,18 @@ class ContactEveryoneSoap(Soap):
send_profiles = self.ProfileListBuilder().string(
context={'recipients': [{'to': to} for to in recipients]})
message.sendProfiles = send_profiles
# XXX: handle webfault and other problems
try:
client.service.sendMessage(message)
except Exception, e:
logging.getLogger(__name__).exception('failed to send sms')
raise
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):
client = self.get_client()
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
@ -68,9 +74,9 @@ class ContactEveryoneSoap(Soap):
send_profiles = self.ProfileListBuilder().string(
context={'recipients': [{'to': to} for to in recipients]})
message.sendProfiles = send_profiles
# XXX: handle webfault and other problems
try:
return client.service.sendAdvancedMessage(message)
except Exception, e:
logging.getLogger(__name__).exception('failed to send sms')
raise
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]}

133
passerelle/apps/ovh/models.py
View File

@ -1,43 +1,76 @@
import urllib
import logging
import json
import requests
from django.utils.translation import ugettext_lazy as _
from django.db import models
from passerelle.utils.jsonresponse import APIError
from passerelle.base.models import BaseResource
from passerelle.sms import SMSGatewayMixin
logger = logging.getLogger('passerelle.apps.ovh')
class OVHError(Exception):
pass
class OVHSMSGateway(BaseResource, SMSGatewayMixin):
URL = 'https://www.ovh.com/cgi-bin/sms/http2sms.cgi'
MESSAGES_CLASSES = (
(0, _('Message are directly shown to users on phone screen '
'at reception. The message is never stored, neither in the '
'phone memory nor in the SIM card. It is deleted as '
'soon as the user validate the display.')),
(1, _('Messages are stored in the phone memory, or in the '
'SIM card if the memory is full. ')),
(2, _('Messages are stored in the SIM card.')),
(3, _('Messages are stored in external storage like a PDA or '
'a PC.')),
(0, _('Message are directly shown to users on phone screen '
'at reception. The message is never stored, neither in the '
'phone memory nor in the SIM card. It is deleted as '
'soon as the user validate the display.')),
(1, _('Messages are stored in the phone memory, or in the '
'SIM card if the memory is full. ')),
(2, _('Messages are stored in the SIM card.')),
(3, _('Messages are stored in external storage like a PDA or '
'a PC.')),
)
account = models.CharField(max_length=64)
username = models.CharField(max_length=64)
password = models.CharField(max_length=64)
msg_class = models.IntegerField(choices=MESSAGES_CLASSES,
default=1, verbose_name=_('message class'))
msg_class = models.IntegerField(choices=MESSAGES_CLASSES, default=1,
verbose_name=_('message class'))
credit_threshold_alert = models.PositiveIntegerField(default=100)
default_country_code = models.CharField(max_length=3, default=u'33')
credit_left = models.PositiveIntegerField(default=0)
# FIXME: add regexp field, to check destination and from format
TEST_DEFAULTS = {
'create_kwargs': {
'account': '1234',
'username': 'john',
'password': 'doe',
},
'test_vectors': [
{
'response': '',
'result': {
'err': 1,
'err_desc': 'OVH error: bad JSON response, \'\', '
'No JSON object could be decoded',
}
},
{
'response': {
'status': 100,
'creditLeft': 47,
'SmsIds': [1234],
},
'result': {
'err': 0,
'data': {
'credit_left': 47.0,
'ovh_result': {
'SmsIds': [1234],
'creditLeft': 47,
'status': 100
},
'sms_ids': [1234],
'warning': 'credit level too low for ovhsmsgateway: 47.0 (threshold 100)',
}
}
}
],
}
class Meta:
verbose_name = 'OVH'
db_table = 'sms_ovh'
@ -54,35 +87,39 @@ class OVHSMSGateway(BaseResource, SMSGatewayMixin):
text = unicode(text).encode('utf-8')
to = ','.join(destinations)
params = {
'account': self.account.encode('utf-8'),
'login': self.username.encode('utf-8'),
'password': self.password.encode('utf-8'),
'from': sender.encode('utf-8'),
'to': to,
'message': text,
'contentType': 'text/json',
'class': self.msg_class,
'account': self.account.encode('utf-8'),
'login': self.username.encode('utf-8'),
'password': self.password.encode('utf-8'),
'from': sender.encode('utf-8'),
'to': to,
'message': text,
'contentType': 'text/json',
'class': self.msg_class,
}
ret = {}
try:
stream = urllib.urlopen('%s?%s' % (self.URL, urllib.urlencode(params)))
except Exception, e:
logger.error('unable to urlopen %s: %s', self.URL, e)
raise OVHError('unable to urlopen %s: %s' % (self.URL, e))
response = requests.post(self.URL, data=params)
except requests.RequestException as e:
raise APIError('OVH error: POST failed, %s' % e)
else:
result = json.loads(stream.read())
if 100 <= result['status'] < 200:
credit_left = float(result['creditLeft'])
# update credit left
OVHSMSGateway.objects.filter(id=self.id) \
.update(credit_left=credit_left)
if credit_left < self.credit_threshold_alert:
logger.error(
'credit level too low for %s: %s (threshold %s)',
self.slug, credit_left, self.credit_threshold_alert)
ret['credit_left'] = credit_left
ret['ovh_result'] = result
return ret
try:
result = response.json()
except ValueError as e:
raise APIError('OVH error: bad JSON response, %r, %s' % (response.content, e))
else:
logger.error('OVH error: %r', result)
raise OVHError('OVH error: %r' % result)
if not isinstance(result, dict):
raise APIError('OVH error: bad JSON response %r, it should be a dictionnary' %
result)
if 100 <= result['status'] < 200:
ret = {}
credit_left = float(result['creditLeft'])
# update credit left
OVHSMSGateway.objects.filter(id=self.id).update(credit_left=credit_left)
if credit_left < self.credit_threshold_alert:
ret['warning'] = ('credit level too low for %s: %s (threshold %s)' %
(self.slug, credit_left, self.credit_threshold_alert))
ret['credit_left'] = credit_left
ret['ovh_result'] = result
ret['sms_ids'] = result.get('SmsIds', [])
return ret
else:
raise APIError('OVH error: %r' % result)

72
passerelle/apps/oxyd/models.py
View File

@ -1,19 +1,11 @@
import urllib
import urllib2
import logging
import requests
from django.db import models
from django.core.urlresolvers import reverse
from passerelle.utils.jsonresponse import APIError
from passerelle.base.models import BaseResource
from passerelle.sms import SMSGatewayMixin
logger = logging.getLogger('passerelle.apps.oxyd')
class OxydError(Exception):
pass
class OxydSMSGateway(BaseResource, SMSGatewayMixin):
username = models.CharField(max_length=64)
@ -21,6 +13,34 @@ class OxydSMSGateway(BaseResource, SMSGatewayMixin):
default_country_code = models.CharField(max_length=3, default=u'33')
# FIXME: add regexp field, to check destination and from format
TEST_DEFAULTS = {
'create_kwargs': {
'username': 'john',
'password': 'doe',
},
'test_vectors': [
{
'response': '',
'result': {
'err': 1,
'err_desc': 'OXYD error: some destinations failed',
'data': [
['33688888888', "OXYD error: received content '' instead of 200"],
['33677777777', "OXYD error: received content '' instead of 200"],
],
}
},
{
'response': '200',
'result': {
'err': 0,
'data': None,
}
}
],
}
URL = 'http://sms.oxyd.fr/send.php'
class Meta:
verbose_name = 'Oxyd'
db_table = 'sms_oxyd'
@ -32,29 +52,29 @@ class OxydSMSGateway(BaseResource, SMSGatewayMixin):
def send_msg(self, text, sender, destinations):
"""Send a SMS using the Oxyd provider"""
# unfortunately it lacks a batch API...
destinations = self.clean_numbers(destinations,
self.default_country_code, prefix='')
destinations = self.clean_numbers(destinations, self.default_country_code, prefix='')
results = []
for dest in destinations:
params = urllib.urlencode({
params = {
'id': self.username,
'pass': self.password,
'num': number,
'num': dest,
'sms': text.encode('utf-8'),
'flash': '0'
})
'flash': '0',
}
try:
r = urllib2.urlopen('http://sms.oxyd.fr/send.php', params)
except Exception, e:
# XXX: add proper handling of errors
logger.error('urlopen oxyd.fr failed : %s', e)
raise OxydError('urlopen oxyd.fr failed : %s' % e)
r = requests.post(self.URL, params)
except requests.RequestException as e:
results.append('OXYD error: POST failed, %s' % e)
else:
line = r.read()
code = line.split()[0]
code = r.content and r.content.split()[0]
if code != '200':
logger.error('OXYD error: %r', line)
raise OxydError('OXYD error: %r' % line)
r.close()
results.append('OXYD error: received content %r instead of 200' % r.content)
else:
results.append(0)
if any(results):
raise APIError('OXYD error: some destinations failed', data=zip(destinations, results))
return None
def get_sms_left(self, type='standard'):
raise NotImplementedError

26
passerelle/sms/__init__.py
View File

@ -4,6 +4,8 @@ import re
from django.utils.translation import ugettext_lazy as _
from passerelle.utils.api import endpoint
from passerelle.utils.jsonresponse import APIError
class SMSGatewayMixin(object):
category = _('SMS Providers')
@ -16,7 +18,7 @@ class SMSGatewayMixin(object):
# really unfortunate.
number = ''.join(re.findall('[0-9]', dest))
if dest.startswith('+'):
pass # it already is fully qualified
pass # it already is fully qualified
elif number.startswith('00'):
# assumes 00 is international access code, remove it
number = number[2:]
@ -28,16 +30,18 @@ class SMSGatewayMixin(object):
@endpoint('json-api', perm='can_send_messages', methods=['post'])
def send(self, request, *args, **kwargs):
data = json.loads(request.body)
assert isinstance(data, dict), 'JSON payload is not a dict'
assert 'message' in data, 'missing "message" in JSON payload'
assert 'from' in data, 'missing "from" in JSON payload'
assert 'to' in data, 'missing "to" in JSON payload'
assert isinstance(data['message'], unicode), 'message is not a string'
assert isinstance(data['from'], unicode), 'from is not a string'
assert all(map(lambda x: isinstance(x, unicode), data['to'])), \
try:
data = json.loads(request.body)
assert isinstance(data, dict), 'JSON payload is not a dict'
assert 'message' in data, 'missing "message" in JSON payload'
assert 'from' in data, 'missing "from" in JSON payload'
assert 'to' in data, 'missing "to" in JSON payload'
assert isinstance(data['message'], unicode), 'message is not a string'
assert isinstance(data['from'], unicode), 'from is not a string'
assert all(map(lambda x: isinstance(x, unicode), data['to'])), \
'to is not a list of strings'
except (ValueError, AssertionError) as e:
raise APIError('Payload error: %s' % e)
logging.info('sending message %r to %r with sending number %r',
data['message'], data['to'], data['from'])
self.send_msg(data['message'], data['from'], data['to'])
return {'err': 0}
return self.send_msg(data['message'], data['from'], data['to'])

20
tests/test_api_access.py
View File

@ -52,9 +52,9 @@ def test_access_with_good_signature(setup):
url = signature.sign_url(endpoint_url + '?orig=eservices', '12345')
# for empty payload the connector returns 500 with
# {"err_desc": "missing \"message\" in JSON payload"}
resp = app.post_json(url, {}, status=500)
resp = app.post_json(url, {})
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'missing "message" in JSON payload'
assert resp.json['err_desc'] == 'Payload error: missing "message" in JSON payload'
def test_access_http_auth(setup):
app, oxyd = setup
@ -75,9 +75,9 @@ def test_access_http_auth(setup):
app.authorization = ('Basic', (username, password))
endpoint_url = reverse('generic-endpoint',
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
resp = app.post_json(endpoint_url, {}, status=500)
resp = app.post_json(endpoint_url, {})
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'missing "message" in JSON payload'
assert resp.json['err_desc'] == 'Payload error: missing "message" in JSON payload'
def test_access_apikey(setup):
app, oxyd = setup
@ -97,9 +97,9 @@ def test_access_apikey(setup):
params = {'message': 'test'}
endpoint_url = reverse('generic-endpoint',
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
resp = app.post_json(endpoint_url + '?apikey=' + password , params, status=500)
resp = app.post_json(endpoint_url + '?apikey=' + password , params)
resp.json['err'] == 1
assert resp.json['err_desc'] == 'missing "from" in JSON payload'
assert resp.json['err_desc'] == 'Payload error: missing "from" in JSON payload'
resp = app.post_json(endpoint_url + '?apikey=' + password[:3] , params, status=403)
resp.json['err'] == 1
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
@ -119,9 +119,9 @@ def test_access_apiuser_with_no_key(setup):
params = {'message': 'test', 'from': 'test api'}
endpoint_url = reverse('generic-endpoint',
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
resp = app.post_json(endpoint_url, params, status=500)
resp = app.post_json(endpoint_url, params)
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'missing "to" in JSON payload'
assert resp.json['err_desc'] == 'Payload error: missing "to" in JSON payload'
def test_access_apiuser_with_ip_restriction(setup):
app, oxyd = setup
@ -148,6 +148,6 @@ def test_access_apiuser_with_ip_restriction(setup):
endpoint_url = reverse('generic-endpoint',
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
resp = app.post_json(endpoint_url, {},
extra_environ=[('REMOTE_ADDR', authorized_ip)], status=500)
extra_environ=[('REMOTE_ADDR', authorized_ip)])
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'missing "message" in JSON payload'
assert resp.json['err_desc'] == 'Payload error: missing "message" in JSON payload'

51
tests/test_sms.py
View File

@ -1,11 +1,62 @@
import pytest
from django.contrib.contenttypes.models import ContentType
from passerelle.base.models import ApiUser, AccessRight
from passerelle.sms import SMSGatewayMixin
import utils
klasses = SMSGatewayMixin.__subclasses__()
def test_clean_numbers():
assert SMSGatewayMixin.clean_numbers(['+ 33 12'], '33') == ['+3312']
assert SMSGatewayMixin.clean_numbers(['0 0 33 12'], '33') == ['+3312']
assert SMSGatewayMixin.clean_numbers(['0 12'], '33') == ['+3312']
def test_clean_numbers_no_prefix():
assert SMSGatewayMixin.clean_numbers(['+ 33 12'], '33', prefix='') == ['3312']
assert SMSGatewayMixin.clean_numbers(['0 0 33 12'], '33', prefix='') == ['3312']
assert SMSGatewayMixin.clean_numbers(['0 12'], '33', prefix='') == ['3312']
@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):
path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
result = app.post_json(path, {})
assert result.json['err'] == 1
assert result.json['err_desc'].startswith('Payload error: ')
payload = {
'message': 'hello',
'from': '+33699999999',
'to': ['+33688888888', '+33677777777'],
}
for test_vector in getattr(connector, 'TEST_DEFAULTS', {}).get('test_vectors', []):
with utils.mock_url(connector.URL, test_vector['response']):
result = app.post_json(path, payload)
print result.json
for key, value in test_vector['result'].iteritems():
assert key in result.json
assert result.json[key] == value

13
tests/utils.py
View File

@ -1,6 +1,8 @@
import json
import urlparse
import mock
import httmock
from django.contrib.contenttypes.models import ContentType
from django.core.urlresolvers import reverse
@ -26,3 +28,14 @@ class FakedResponse(mock.Mock):
def json(self):
return json.loads(self.content)
def mock_url(url, response):
parsed = urlparse.urlparse(url)
if not isinstance(response ,str):
response = json.dumps(response)
@httmock.urlmatch(netloc=parsed.netloc, path=parsed.path)
def mocked(url, request):
return response
return httmock.HTTMock(mocked)

3
tox.ini
View File

@ -8,6 +8,7 @@ usedevelop =
setenv =
DJANGO_SETTINGS_MODULE=passerelle.settings
PASSERELLE_SETTINGS_FILE=tests/settings.py
fast: FAST=--nomigrations
coverage: COVERAGE=--junitxml=test_results.xml --cov-report xml --cov=passerelle/ --cov-config .coveragerc
deps =
django17: django>1.7,<1.8
@ -25,5 +26,5 @@ deps =
django-webtest
commands =
./getmagic.sh
py.test {env:COVERAGE:} {posargs:tests/}
py.test {env:FAST:} {env:COVERAGE:} {posargs:tests/}
pylint: ./pylint.sh passerelle/