Produits Entr'ouvert » Passerelle

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

# Generated by Django 1.11.18 on 2020-07-30 08:47

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('ovh', '0008_ovhsmsgateway_max_message_length'),

    ]

    operations = [

        migrations.AddField(

            model_name='ovhsmsgateway',

            name='application_key',

            field=models.CharField(blank=True, help_text='Random token obtained from OVH.', max_length=16, verbose_name='Application key'),

        ),

        migrations.AddField(

            model_name='ovhsmsgateway',

            name='application_secret',

            field=models.CharField(blank=True, help_text='Obtained at the same time as "Application key".', max_length=32, verbose_name='Application secret'),

        ),

        migrations.AddField(

            model_name='ovhsmsgateway',

            name='consumer_key',

            field=models.CharField(blank=True, help_text='Obtained at the same time as "Application key".', max_length=32, verbose_name='Consumer key'),

        ),

        migrations.AlterField(

            model_name='ovhsmsgateway',

            name='account',

            field=models.CharField(help_text='Account identifier, such as sms-XXXXXX-1.', max_length=64, verbose_name='Account'),

        ),

        migrations.AlterField(

            model_name='ovhsmsgateway',

            name='password',

            field=models.CharField(blank=True, help_text='Password for legacy API. This field is obsolete once keys and secret fields below are filled.', max_length=64, verbose_name='Password (deprecated)'),

        ),

        migrations.AlterField(

            model_name='ovhsmsgateway',

            name='username',

            field=models.CharField(help_text='API user created on the SMS account.', max_length=64, verbose_name='Username'),

        ),

    ]

import hashlib

import json

import time

    documentation_url = 'https://doc-publik.entrouvert.com/admin-fonctionnel/les-tutos/configuration-envoi-sms/'

    API_URL = 'https://eu.api.ovh.com/1.0/sms/%(serviceName)s/users/%(login)s/jobs/'

    NEW_MESSAGES_CLASSES = ['flash', 'phoneDisplay', 'sim', 'toolkit']

    account = models.CharField(

        verbose_name=_('Account'), max_length=64, help_text=_('Account identifier, such as sms-XXXXXX-1.')

    )

    application_key = models.CharField(

        verbose_name=_('Application key'),

        max_length=16,

        blank=True,

        help_text=_('Random token obtained from OVH.'),

    )

    application_secret = models.CharField(

        verbose_name=_('Application secret'),

        max_length=32,

        blank=True,

        help_text=_('Obtained at the same time as "Application key".'),

    )

    consumer_key = models.CharField(

        verbose_name=_('Consumer key'),

        max_length=32,

        blank=True,

        help_text=_('Obtained at the same time as "Application key".'),

    )

    username = models.CharField(

        verbose_name=_('Username'),

        max_length=64,

        help_text=_('API user created on the SMS account.'),

    )

    password = models.CharField(

        verbose_name=_('Password (deprecated)'),

        max_length=64,

        blank=True,

        help_text=_(

            'Password for legacy API. This field is obsolete once keys and secret fields below are filled.'

        ),

    )

        if not (self.application_key and self.consumer_key and self.application_secret):

            return self.send_msg_legacy(text, sender, destinations, **kwargs)

        url = self.API_URL % {'serviceName': self.account, 'login': self.username}

        body = {

            'sender': sender,

            'receivers': destinations,

            'message': text,

            'class': self.NEW_MESSAGES_CLASSES[self.msg_class],

        }

        if not kwargs['stop']:

            body.update({'noStopClause': 1})

        # sign request

        now = str(int(time.time()))

        signature = hashlib.sha1()

        to_sign = "+".join((self.application_secret, self.consumer_key, 'POST', url, json.dumps(body), now))

        signature.update(to_sign.encode())

        headers = {

            'X-Ovh-Application': self.application_key,

            'X-Ovh-Consumer': self.consumer_key,

            'X-Ovh-Timestamp': now,

            'X-Ovh-Signature': "$1$" + signature.hexdigest(),

        }

        try:

            response = self.requests.post(url, headers=headers, json=body)

        except requests.RequestException as e:

            raise APIError('OVH error: POST failed, %s' % e)

        else:

            try:

                result = response.json()

            except ValueError as e:

                raise APIError('OVH error: bad JSON response')

        try:

            response.raise_for_status()

        except requests.RequestException as e:

            raise APIError('OVH error: %s "%s"' % (e, result))

        ret = {}

        credits_removed = result['totalCreditsRemoved']

        # update credit left

        self.credit_left -= credits_removed

        if self.credit_left < 0:

            self.credit_left = 0

        self.save()

        if self.credit_left < self.credit_threshold_alert:

            ret['warning'] = 'credit level too low for %s: %s (threshold %s)' % (

                self.slug,

                self.credit_left,

                self.credit_threshold_alert,

            )

        ret['credit_left'] = self.credit_left

        ret['ovh_result'] = result

        ret['sms_ids'] = result.get('ids', [])

        return ret

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

        """Send a SMS using the HTTP2 endpoint"""

        if not self.password:

            raise APIError('Improperly configured, empty keys or password fields.')

def test_ovh_new_api(app, freezer):

    connector = OVHSMSGateway.objects.create(

        slug='ovh', account='sms-test42', username='john',

        application_key='RHrTdU2oTsrVC0pu',

        application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',

        consumer_key='iF0zi0MJrbjNcI3hvuvwkhNk8skrigxz'

    )

    api = ApiUser.objects.create(username='apiuser')

    obj_type = ContentType.objects.get_for_model(connector)

    # no access check

    AccessRight.objects.create(codename='can_send_messages', apiuser=api, resource_type=obj_type,

                               resource_pk=connector.pk)

    payload = {

        'message': 'hello',

        'from': '+33699999999',

        'to': ['+33688888888', '+33677777777'],

    }

    # register job

    freezer.move_to('2019-01-01 00:00:00')

    path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)

    result = app.post_json(path, params=payload)

    assert result.json['err'] == 0

    job_id = Job.objects.get(status='registered').id

    # perform job

    freezer.move_to('2019-01-01 01:00:03')

    resp = {

        'validReceivers': ['+33688888888', '+33677777777'],

        'totalCreditsRemoved': 1,

        'ids': [241615100],

        'invalidReceivers': []

    }

    url = connector.API_URL % {'serviceName': 'sms-test42', 'login': 'john'}

    with utils.mock_url(url, resp, 200) as mocked:

        connector.jobs()

    job = Job.objects.get(id=job_id)

    assert job.status == 'completed'

    request = mocked.handlers[0].call['requests'][0]

    assert 'X-Ovh-Signature' in request.headers

    @httmock.remember_called