0001-ovh-update-credit-left-42921.patch

Valentin Deniaud, 08 octobre 2020 11:44

Voir les différences: en ligne côte à côte

Subject: [PATCH] ovh: update credit left (#42921)

 .../ovh/migrations/0010_auto_20201008_1126.py | 20 +++++++
 passerelle/apps/ovh/models.py                 | 60 +++++++++++++------
 .../templates/ovh/ovhsmsgateway_detail.html   | 15 +++++
 tests/test_sms.py                             | 46 ++++++++++++--
 4 files changed, 118 insertions(+), 23 deletions(-)
 create mode 100644 passerelle/apps/ovh/migrations/0010_auto_20201008_1126.py
 create mode 100644 passerelle/apps/ovh/templates/ovh/ovhsmsgateway_detail.html

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.18 on 2020-10-08 09:26
     from __future__ import unicode_literals
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('ovh', '0009_auto_20200730_1047'),
+        ]
         operations = [
             migrations.AlterField(
                 model_name='ovhsmsgateway',
                 name='credit_left',
                 field=models.PositiveIntegerField(default=0, editable=False, verbose_name='Credit left'),
             ),
+        ]

     import json
     import requests
     import time
     from urllib.parse import urljoin
     from django.db import models
     from django.utils.encoding import force_text
-...
     class OVHSMSGateway(SMSResource):
         documentation_url = 'https://doc-publik.entrouvert.com/admin-fonctionnel/les-tutos/configuration-envoi-sms/'
         hide_description_fields = ['account']
         API_URL = 'https://eu.api.ovh.com/1.0/sms/%(serviceName)s/users/%(login)s/jobs/'
         hide_description_fields = ['account', 'credit_left']
         API_URL = 'https://eu.api.ovh.com/1.0/sms/%(serviceName)s/'
         URL = 'https://www.ovh.com/cgi-bin/sms/http2sms.cgi'
         MESSAGES_CLASSES = (
             (0, _('Message are directly shown to users on phone screen '
-...
                                         verbose_name=_('Message class'))
         credit_threshold_alert = models.PositiveIntegerField(verbose_name=_('Credit alert threshold'),
                                                              default=100)
         credit_left = models.PositiveIntegerField(verbose_name=_('Credit left'), default=0)
         credit_left = models.PositiveIntegerField(verbose_name=_('Credit left'), default=0, editable=False)
         TEST_DEFAULTS = {
             'create_kwargs': {
-...
             verbose_name = 'OVH'
             db_table = 'sms_ovh'
         def send_msg(self, text, sender, destinations, **kwargs):
             if not (self.application_key and self.consumer_key and self.application_secret):
                 return self.send_msg_legacy(text, sender, destinations, **kwargs)
         @property
         def uses_new_api(self):
             return self.application_key and self.consumer_key and self.application_secret
         def request(self, method, endpoint, **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})
             url = urljoin(url, endpoint)
             # sign request
             body = json.dumps(kwargs['json']) if 'json' in kwargs else ''
             now = str(int(time.time()))
             signature = hashlib.sha1()
             to_sign = "+".join((self.application_secret, self.consumer_key, 'POST', url, json.dumps(body), now))
             to_sign = "+".join((self.application_secret, self.consumer_key, method.upper(), url, body, now))
             signature.update(to_sign.encode())
             headers = {
-...
+            }
             try:
                 response = self.requests.post(url, headers=headers, json=body)
                 response = self.requests.request(method, url, headers=headers, **kwargs)
             except requests.RequestException as e:
                 raise APIError('OVH error: POST failed, %s' % e)
             else:
-...
                 response.raise_for_status()
             except requests.RequestException as e:
                 raise APIError('OVH error: %s "%s"' % (e, result))
             return result
         def send_msg(self, text, sender, destinations, **kwargs):
             if not self.uses_new_api:
                 return self.send_msg_legacy(text, sender, destinations, **kwargs)
             body = {
                 'sender': sender,
                 'receivers': destinations,
                 'message': text,
                 'class': self.NEW_MESSAGES_CLASSES[self.msg_class],
+            }
             if not kwargs['stop']:
                 body.update({'noStopClause': 1})
             result = self.request('post', 'jobs/', json=body)
             ret = {}
             credits_removed = result['totalCreditsRemoved']
-...
             self.credit_left -= credits_removed
             if self.credit_left < 0:
                 self.credit_left = 0
             self.save()
             self.save(update_credit=False)
             if self.credit_left < self.credit_threshold_alert:
                 ret['warning'] = 'credit level too low for %s: %s (threshold %s)' % (
                     self.slug,
-...
             return ret
         def update_credit_left(self):
             result = self.request('get', endpoint='')
             self.credit_left = result['creditsLeft']
             self.save(update_credit=False)
         def hourly(self):
             super().hourly()
             self.update_credit_left()
         def save(self, *args, update_credit=True, **kwargs):
             super().save(*args, **kwargs)
             if update_credit:
                 self.add_job('update_credit_left')
         def send_msg_legacy(self, text, sender, destinations, **kwargs):
             """Send a SMS using the HTTP2 endpoint"""
             if not self.password:

     {% extends "passerelle/manage/messages_service_view.html" %}
     {% load i18n passerelle %}
     {% block description %}
     {{ block.super }}
     {% if object.uses_new_api %}
     <p>
     {% if object.credit_left %}
     <b>{% trans "Credit left:" %}</b>  {{ object.credit_left }}
     {% else %}
     <b>{% trans "There is no credit left." %}</b>
     {% endif %}
     </p>
     {% endif %}
     {% endblock %}

         test_vectors = getattr(connector, 'TEST_DEFAULTS', {}).get('test_vectors', [])
         total = len(test_vectors)
         nb_failed = 0
         assert Job.objects.count() == 0
         assert Job.objects.filter(method_name='send_job').count() == 0
         for test_vector in test_vectors:
             # register job
             freezer.move_to('2019-01-01 00:00:00')
             result = app.post_json(path, params=payload)
             assert result.json['err'] == 0
             job_id = Job.objects.get(status='registered').id
             job_id = Job.objects.get(method_name='send_job', status='registered').id
             # perform job
             freezer.move_to('2019-01-01 01:00:03')
-...
                 nb_failed += 1
             else:
                 assert job.status == 'completed'
         assert Job.objects.count() == total
         assert Job.objects.filter(method_name='send_job').count() == total
         assert SMSLog.objects.count() == total - nb_failed
-...
     def test_ovh_new_api(app, freezer):
         connector = OVHSMSGateway.objects.create(
             slug='ovh', account='sms-test42', username='john',
             slug='ovh', account='sms-test42',
             application_key='RHrTdU2oTsrVC0pu',
             application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
             consumer_key='iF0zi0MJrbjNcI3hvuvwkhNk8skrigxz'
-...
         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
         job_id = Job.objects.get(method_name='send_job', status='registered').id
         # perform job
         freezer.move_to('2019-01-01 01:00:03')
-...
             'ids': [241615100],
             'invalidReceivers': []
+        }
         url = connector.API_URL % {'serviceName': 'sms-test42', 'login': 'john'}
         base_url = connector.API_URL % {'serviceName': 'sms-test42'}
         url = base_url + 'jobs/'
         with utils.mock_url(url, resp, 200) as mocked:
             connector.jobs()
         job = Job.objects.get(id=job_id)
-...
         request = mocked.handlers[0].call['requests'][0]
         assert 'X-Ovh-Signature' in request.headers
     def test_ovh_new_api_credit(app, freezer):
         connector = OVHSMSGateway.objects.create(
             slug='ovh', account='sms-test42',
             application_key='RHrTdU2oTsrVC0pu',
             application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
             consumer_key='iF0zi0MJrbjNcI3hvuvwkhNk8skrigxz'
+        )
         manager_url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
         resp = app.get(manager_url)
         assert 'no credit left' in resp.text
         # a job to update credit was added on connector creation
         resp = {
             'creditsLeft': 123,
+        }
         ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}
         with utils.mock_url(ovh_url, resp, 200) as mocked:
             connector.jobs()
         assert connector.credit_left == 123
         resp = app.get(manager_url)
         assert '123' in resp.text
         # hourly update
         resp = {
             'creditsLeft': 456,
+        }
         with utils.mock_url(ovh_url, resp, 200) as mocked:
             connector.hourly()
         assert connector.credit_left == 456
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0001-ovh-update-credit-left-42921.patch