Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.26 on 2022-09-21 08:39

import django.contrib.postgres.fields

import django.core.validators

from django.db import migrations, models

import passerelle.sms.models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='SMSFactorSMSGateway',

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

                (

                    'default_country_code',

                    models.CharField(

                        default='33',

                        max_length=3,

                        validators=[

                            django.core.validators.RegexValidator(

                                '^[0-9]*$', 'The country must only contain numbers'

                            )

                        ],

                        verbose_name='Default country code',

                    ),

                ),

                (

                    'default_trunk_prefix',

                    models.CharField(

                        default='0',

                        max_length=2,

                        validators=[

                            django.core.validators.RegexValidator(

                                '^[0-9]*$', 'The trunk prefix must only contain numbers'

                            )

                        ],

                        verbose_name='Default trunk prefix',

                    ),

                ),

                (

                    'max_message_length',

                    models.IntegerField(

                        default=2000,

                        help_text='Messages over this limit will be truncated.',

                        verbose_name='Maximum message length',

                    ),

                ),

                (

                    'authorized',

                    django.contrib.postgres.fields.ArrayField(

                        base_field=models.CharField(

                            choices=[

                                ('fr-metro', 'France mainland (+33 [67])'),

                                ('fr-domtom', 'France DOM/TOM (+262, etc.)'),

                                ('be', 'Belgian (+32 4[5-9]) '),

                                ('all', 'All'),

                            ],

                            max_length=32,

                            null=True,

                        ),

                        default=passerelle.sms.models.authorized_default,

                        size=None,

                        verbose_name='Authorized Countries',

                    ),

                ),

                ('auth_token', models.CharField(max_length=255, verbose_name='Auth Token')),

                (

                    'credit_threshold_alert',

                    models.PositiveIntegerField(default=500, verbose_name='Credit alert threshold'),

                ),

                (

                    'credit_left',

                    models.PositiveIntegerField(default=0, editable=False, verbose_name='Credit left'),

                ),

                (

                    'alert_emails',

                    django.contrib.postgres.fields.ArrayField(

                        base_field=models.EmailField(blank=True, max_length=254),

                        blank=True,

                        null=True,

                        size=None,

                        verbose_name='Email addresses list to send credit alerts to, separated by comma',

                    ),

                ),

                ('credit_alert_timestamp', models.DateTimeField(editable=False, null=True)),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_smsfactorsmsgateway_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'SMS Factor',

                'db_table': 'sms_factor',

            },

        ),

    ]

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

import logging

import urllib.parse

import requests

from django.conf import settings

from django.contrib.postgres.fields import ArrayField

from django.core.mail import send_mail

from django.db import models

from django.template.loader import render_to_string

from django.utils import timezone

from django.utils.translation import gettext_lazy as _

from passerelle.sms.models import SMSResource

from passerelle.utils.jsonresponse import APIError

class SMSFactorSMSGateway(SMSResource):

    auth_token = models.CharField(verbose_name=_('Auth Token'), max_length=255)

    credit_threshold_alert = models.PositiveIntegerField(

        verbose_name=_('Credit alert threshold'), default=500

    )

    credit_left = models.PositiveIntegerField(verbose_name=_('Credit left'), default=0, editable=False)

    alert_emails = ArrayField(

        models.EmailField(blank=True),

        blank=True,

        null=True,

        verbose_name=_('Email addresses list to send credit alerts to, separated by comma'),

    )

    credit_alert_timestamp = models.DateTimeField(null=True, editable=False)

    # unecessary field

    allow_premium_rate = None

    class Meta:

        verbose_name = 'SMS Factor'

        db_table = 'sms_factor'

    TEST_DEFAULTS = {

        'create_kwargs': {

            'auth_token': 'yyy',

            'credit_threshold_alert': 1000,

        },

        'test_vectors': [

            {

                'status_code': 200,

                'response': {

                    "status": -7,

                    "message": "Erreur de données",

                    "details": "Texte du message introuvable",

                },

                'result': {

                    'err': 1,

                    'err_desc': 'SMS Factor error: some destinations failed',

                    'data': [

                        ['33688888888', "Texte du message introuvable"],

                        ['33677777777', "Texte du message introuvable"],

                    ],

                },

            },

            {

                'status_code': 200,

                'response': {

                    "status": 1,

                    "message": "OK",

                    "ticket": "14672468",

                    "cost": 2,

                    "credits": 642,

                    "total": 2,

                    "sent": 2,

                    "blacklisted": 0,

                    "duplicated": 0,

                    "invalid": 0,

                    "npai": 0,

                },

                'result': {

                    'err': 0,

                    'data': {

                        "status": 1,

                        "message": "OK",

                        "ticket": "14672468",

                        "cost": 2,

                        "credits": 642,

                        "total": 2,

                        "sent": 2,

                        "blacklisted": 0,

                        "duplicated": 0,

                        "invalid": 0,

                        "npai": 0,

                    },

                },

            },

        ],

    }

    URL = 'https://api.smsfactor.com'

    def request(self, method, endpoint, **kwargs):

        url = urllib.parse.urljoin(self.URL, endpoint)

        headers = {

            "Authorization": f"Bearer {self.auth_token}",

            "Accept": "application/json",

        }

        try:

            response = self.requests.request(method, url, headers=headers, **kwargs)

        except requests.RequestException as e:

            raise APIError('SMS Factor: request failed, %s' % e)

        else:

            try:

                result = response.json()

            except ValueError:

                raise APIError('SMS Factor: bad JSON response')

        try:

            response.raise_for_status()

        except requests.RequestException as e:

            raise APIError('SMS Factor: %s "%s"' % (e, result))

        return result

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

        """Send a SMS using the SMS Factor provider"""

        # from https://dev.smsfactor.com/en/api/sms/send/send-single

        # and https://dev.smsfactor.com/en/api/sms/send/send-simulate

        # set destinations phone number in E.164 format (without the + prefix)

        # [country code][phone number including area code]

        endpoint = 'send'

        numbers = []

        for dest in destinations:

            numbers.append(dest[2:])

        destinations = numbers

        results = []

        for dest in destinations:

            params = {

                'sender': sender,

                'text': text,

                'to': dest,

                'pushtype': 'alert' if kwargs.get('stop', False) else 'marketing',

            }

            data = self.request('get', endpoint, params=params)

            logging.info('SMS Factor answered with %s', data)

            results.append(data)

        errors = [f'SMS Factor error: {r["status"]}: {r["message"]}' for r in results if r['status'] != 1]

        consumed_credits = None

        try:

            self.credit_left = results[-1]['credits']

            consumed_credits = sum((r['cost'] for r in results))

            self.save(update_fields=['credit_left'])

        except KeyError:

            # no credits key, there was probably an error with the request

            pass

        if any(errors):

            raise APIError('SMS Factor error: some destinations failed', data=errors)

        return consumed_credits

    def update_credit_left(self):

        result = self.request('get', endpoint='credits')

        self.credit_left = result['credits']

        self.save(update_fields=['credit_left'])

    def send_credit_alert_if_needed(self):

        if self.credit_left >= self.credit_threshold_alert:

            return

        if self.credit_alert_timestamp and self.credit_alert_timestamp > timezone.now() - datetime.timedelta(

            days=1

        ):

            return  # alerts are sent daily

        ctx = {

            'connector': self,

            'connector_url': urllib.parse.urljoin(settings.SITE_BASE_URL, self.get_absolute_url()),

        }

        subject = render_to_string('smsfactor/credit_alert_subject.txt', ctx).strip()

        body = render_to_string('smsfactor/credit_alert_body.txt', ctx)

        html_body = render_to_string('smsfactor/credit_alert_body.html', ctx)

        send_mail(

            subject,

            body,

            settings.DEFAULT_FROM_EMAIL,

            self.alert_emails,

            html_message=html_body,

        )

        self.credit_alert_timestamp = timezone.now()

        self.save()

        self.logger.warning('credit is too low, alerts were sent to %s', self.alert_emails)

    def hourly(self):

        super().hourly()

        self.update_credit_left()

        self.send_credit_alert_if_needed()

{% extends "emails/body_base.html" %}

{% load i18n %}

{% block content %}

<p>{% trans "Hi," %}</p>

<p>

{% blocktrans trimmed with name=connector.title credit_left=connector.credit_left %}

There are only {{ credit_left }} credits left for connector {{ name }}.

{% endblocktrans %}

</p>

<p>

{% blocktrans trimmed with account=connector.account %}

Please add more credit as soon as possible for your SMS Factor account.

{% endblocktrans %}

</p>

{% with _("View connector page") as button_label %}

{% include "emails/button-link.html" with url=connector_url label=button_label %}

{% endwith %}

{% endblock %}

{% extends "emails/body_base.txt" %}

{% load i18n %}

{% block content %}{% autoescape off %}{% trans "Hi," %}

{% blocktrans trimmed with name=connector.title credit_left=connector.credit_left %}

There are only {{ credit_left }} credits left for connector {{ name }}.

{% endblocktrans %}

{% blocktrans trimmed with account=connector.account %}

Please add more credit as soon as possible for your SMS Factor account..

{% endblocktrans %}

{% trans "View connector page:" %} {{ connector_url }}

{% endautoescape %}

{% endblock %}

{% extends "emails/subject.txt" %}

{% load i18n %}

{% block email-subject %}{% autoescape off %}{% blocktrans trimmed with credit_left=connector.credit_left %}

SMS Factor alert: only {{ credit_left }} credits left

{% endblocktrans %}{% endautoescape %}{% endblock %}

    'passerelle.apps.smsfactor',

from passerelle.apps.smsfactor.models import SMSFactorSMSGateway

@pytest.mark.parametrize('connector', [OVHSMSGateway, SMSFactorSMSGateway], indirect=True)

    with mock.patch.object(connector.__class__, 'send_msg') as send_function:

    with mock.patch.object(connector.__class__, 'send_msg') as send_function:

@pytest.mark.parametrize('connector', [OVHSMSGateway, SMSFactorSMSGateway], indirect=True)

    with mock.patch.object(connector.__class__, 'send_msg') as send_function:

    if isinstance(connector, SMSFactorSMSGateway):

        pytest.skip(f"nostop parameter not supported for {connector}")

@pytest.mark.parametrize('connector', [OVHSMSGateway, SMSFactorSMSGateway], indirect=True)

def test_sms_factor_alert_emails(app, freezer, mailoutbox):

    connector = SMSFactorSMSGateway.objects.create(

        slug='test-sms-factor',

        title='Test SMS Factor',

        auth_token='foo',

        credit_threshold_alert=100,

        credit_left=102,

        alert_emails=['test@entrouvert.org'],

    )

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

    obj_type = ContentType.objects.get_for_model(connector)

    AccessRight.objects.create(

        codename='can_send_messages', apiuser=api, resource_type=obj_type, resource_pk=connector.pk

    )

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

    resp = {'credits': 101}

    url = connector.URL

    with tests.utils.mock_url(url, resp, 200):

        connector.hourly()

    assert len(mailoutbox) == 0

    resp = {'credits': 99}

    url = connector.URL

    with tests.utils.mock_url(url, resp, 200):

        connector.hourly()

    assert len(mailoutbox) == 1

    mail = mailoutbox[0]

    assert mail.recipients() == ['test@entrouvert.org']

    assert mail.subject == 'SMS Factor alert: only 99 credits left'

    for body in (mail.body, mail.alternatives[0][0]):

        assert "SMS Factor" in body

        assert connector.title in body

        assert 'http://localhost/smsfactor/test-sms-factor/' in body

    mailoutbox.clear()

    # alert is sent again daily

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

    resp = {'credits': 99}

    url = connector.URL

    with tests.utils.mock_url(url, resp, 200):

        connector.hourly()

    assert len(mailoutbox) == 0

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

    with tests.utils.mock_url(url, resp, 200):

        connector.hourly()

    assert len(mailoutbox) == 1