Produits Entr'ouvert » Passerelle

# -*- 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,

        ),

    ]

# -*- 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.utils.translation import ugettext_lazy as _

from passerelle.utils.jsonresponse import APIError

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)

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

        verbose_name = _('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)

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

# -*- 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'