Produits Entr'ouvert » Passerelle

# -*- coding: utf-8 -*-

# Generated by Django 1.11.18 on 2020-04-04 11:32

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0018_smslog'),

    ]

    operations = [

        migrations.CreateModel(

            name='OrangeRestSMSGateway',

            fields=[

                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                ('max_message_length', models.IntegerField(default=160, verbose_name='Maximum message length')),

                ('username', models.CharField(max_length=64, verbose_name='Identifiant')),

                ('password', models.CharField(max_length=64, verbose_name='Mot de passe')),

                ('groupname', models.CharField(max_length=64, verbose_name='Groupe')),

                ('default_country_code', models.CharField(default='33', max_length=3, verbose_name='Préfixe pays')),

                ('default_trunk_prefix', models.CharField(default='0', max_length=2, verbose_name='Préfixe supprimé par défaut')),

                ('users', models.ManyToManyField(blank=True, related_name='_orangerestsmsgateway_users_+', related_query_name='+', to='base.ApiUser')),

            ],

            options={

                'verbose_name': 'Orange REST SMS',

                'db_table': 'sms_orangerest',

            },

        ),

    ]

# -*- 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/>.

import json

from django.db import models

from passerelle.base.models import SMSResource

from passerelle.utils.jsonresponse import APIError

class OrangeError(APIError):

    pass

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'

def get_json(response):

    try:

        return response.json()

    except ValueError:

        raise OrangeError('Orange returned Invalid JSON content: %s' % response.text)

def login(requests, username, password):

    headers = {'content-type': 'application/x-www-form-urlencoded'}

    params = {'username': username, 'password': password}

    response = 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(requests, access_token, group_name):

    headers = {'authorization': 'Bearer %s' % access_token}

    response = 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'] == group_name:

            group_id = group['id']

            break

    if not group_id:

        raise APIError('Group name not found: ' + group_name)

    return group_id

def send(requests, access_token, group_id, destinations, message):

    headers = {

        'content-type': 'application/json',

        'authorization': 'Bearer %s' % access_token,

    }

    payload = json.dumps({

        'name': 'Send a SMS from passerelle',

        'msisdns': destinations,

        'smsParam': {

            'encoding': 'GSM7',

            'body': message

        }

    })

    response = requests.post(URL_DIFFUSION % group_id, data=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)

class OrangeRestSMSGateway(SMSResource):

    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)

    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 REST'

        db_table = 'sms_orangerest'

    def send_msg(self, text, sender, destinations, **kwargs):

        '''Send a SMS using the Orange REST'''

        destinations = self.clean_numbers(

            destinations, self.default_country_code, self.default_trunk_prefix)

        access_token = login(self.requests, self.username, self.password)

        group_id = group_id_from_name(self.requests, access_token, self.groupname)

        response = send(self.requests, access_token, group_id, destinations, text)

        return response

    'passerelle.apps.orange_sms',

# 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 requests

import httmock

import pytest

from django.contrib.contenttypes.models import ContentType

from passerelle.apps.orange_sms.models import (

    OrangeRestSMSGateway, OrangeError, login, group_id_from_name, send)

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_login_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_id_from_name_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_send_ok(url, request):

    assert request.headers['authorization'] == 'Bearer my_token'

    assert json.loads(request.body)['smsParam']['body'] == PAYLOAD['message']

    '33688888888' in json.loads(request.body)['msisdns'][0]

    '33677777777' in json.loads(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(

        OrangeRestSMSGateway.objects.create(

            slug='my_connector',

            username='jdoe',

            password='secret',

            groupname='group2'

        ))

def test_login(app, connector):

    with httmock.HTTMock(response_login_ok):

        assert login(requests, 'jdoe', 'secret') == 'my_token'

    # not 200

    with pytest.raises(APIError, match='Bad username or password'):

        with httmock.HTTMock(response_500):

            login(requests, 'jdoe', 'secret')

    # not json

    with pytest.raises(OrangeError, match='Orange returned Invalid JSON content'):

        with httmock.HTTMock(response_invalid_json):

            login(requests, 'jdoe', 'secret')

    # 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):

            login(requests, 'jdoe', 'secret')

def test_group_id_from_name(app, connector):

    with httmock.HTTMock(response_group_id_from_name_ok):

        assert group_id_from_name(requests, 'my_token', 'group2') == 'gid2'

    # no group

    with pytest.raises(APIError, match='Group name not found: group3'):

        with httmock.HTTMock(response_group_id_from_name_ok):

            group_id_from_name(requests, 'my_token', 'group3')

    # not 200

    with pytest.raises(APIError, match='Bad token'):

        with httmock.HTTMock(response_500):

            group_id_from_name(requests, 'my_token', 'group2')

    # not json

    with pytest.raises(OrangeError, match='Orange returned Invalid JSON content'):

        with httmock.HTTMock(response_invalid_json):

            group_id_from_name(requests, 'my_token', 'group2')

def test_send(app, connector):

    with httmock.HTTMock(response_send_ok):

        resp = send(requests, '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):

            send(requests, '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):

            send(requests, '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_login_ok, response_group_id_from_name_ok, response_send_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_login_ok, response_group_id_from_name_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'