entrouvert
/
passerelle
16
0
Fork 0
Code Issues Pull Requests 10 Releases Activity

connecteur send ethic (#81143) #346

Merged
csechet merged 3 commits from wip/81143-connecteur-send-ethic into main 2023-09-19 15:01:36 +02:00
Conversation 2 Commits 3 Files Changed 15 +390 -227
15 changed files with 390 additions and 227 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
passerelle/apps/ovh/models.py
View File

@ -2,25 +2,19 @@ import hashlib
import json
import random
import time
from datetime import timedelta
from urllib.parse import urljoin
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.encoding import force_str
from django.utils.translation import gettext_lazy as _
from passerelle.base.models import SkipJob
from passerelle.sms.models import SMSResource
from passerelle.sms.models import TrackCreditSMSResource
from passerelle.utils.jsonresponse import APIError
class OVHSMSGateway(SMSResource):
class OVHSMSGateway(TrackCreditSMSResource):
documentation_url = (
'https://doc-publik.entrouvert.com/admin-fonctionnel/les-tutos/configuration-envoi-sms/'
)
@ -83,17 +77,6 @@ class OVHSMSGateway(SMSResource):
),
)
msg_class = models.IntegerField(choices=MESSAGES_CLASSES, default=1, verbose_name=_('Message class'))
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)
TEST_DEFAULTS = {
'create_kwargs': {
@ -128,6 +111,17 @@ class OVHSMSGateway(SMSResource):
],
}
TEST_CREDIT_LEFT = {
'create_kwargs': {
'account': 'sms-test42',
'application_key': 'RHrTdU2oTsrVC0pu',
'application_secret': 'CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
'consumer_key': 'iF0zi0MJrbjNcI3hvuvwkhNk8skrigxz',
},
'url': (API_URL % {'serviceName': 'sms-test42'}),
'get_credit_left_payload': lambda x: {'creditsLeft': x},
}
class Meta:
verbose_name = 'OVH'
db_table = 'sms_ovh'
@ -191,39 +185,15 @@ class OVHSMSGateway(SMSResource):
self.save(update_credit=False)
return credits_removed
def check_status(self):
if self.uses_new_api:
super().check_status()
def update_credit_left(self):
result = self.request('get', endpoint='')
self.credit_left = result['creditsLeft']
self.save(update_credit=False)
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() - timedelta(days=1):
return # alerts are sent daily
ctx = {
'connector': self,
'connector_url': urljoin(settings.SITE_BASE_URL, self.get_absolute_url()),
}
subject = render_to_string('ovh/credit_alert_subject.txt', ctx).strip()
body = render_to_string('ovh/credit_alert_body.txt', ctx)
html_body = render_to_string('ovh/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 check_status(self):
if self.uses_new_api:
self.update_credit_left()
self.send_credit_alert_if_needed()
def save(self, *args, update_credit=True, **kwargs):
super().save(*args, **kwargs)
if update_credit and self.uses_new_api:

22
passerelle/apps/ovh/templates/ovh/credit_alert_body.html
View File

@ -1,22 +0,0 @@
{% 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 OVH account {{ account }}.
{% endblocktrans %}
</p>
{% with _("View connector page") as button_label %}
{% include "emails/button-link.html" with url=connector_url label=button_label %}
{% endwith %}
{% endblock %}

17
passerelle/apps/ovh/templates/ovh/credit_alert_body.txt
View File

@ -1,17 +0,0 @@
{% 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 OVH account {{ account }}.
{% endblocktrans %}
{% trans "View connector page:" %} {{ connector_url }}
{% endautoescape %}
{% endblock %}

0
passerelle/apps/sendethic/__init__.py Normal file
View File

116
passerelle/apps/sendethic/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,116 @@
# Generated by Django 3.2.18 on 2023-09-18 17:55
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', '0030_resourcelog_base_resour_appname_298cbc_idx'),
]
operations = [
migrations.CreateModel(
name='SendEthicSMSGateway',
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',
),
),
(
'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)),
('account_id', models.CharField(max_length=255, verbose_name='Account ID')),
('api_key', models.CharField(max_length=255, verbose_name='API Key')),
(
'users',
models.ManyToManyField(
blank=True,
related_name='_sendethic_sendethicsmsgateway_users_+',
related_query_name='+',
to='base.ApiUser',
),
),
],
options={
'verbose_name': 'Sendethic',
'db_table': 'sms_send_ethic',
},
),
]

0
passerelle/apps/sendethic/migrations/__init__.py Normal file
View File

141
passerelle/apps/sendethic/models.py Normal file
View File

@ -0,0 +1,141 @@
# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2022 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 urllib.parse
import requests
from django.db import models
from django.utils.translation import gettext_lazy as _
from passerelle.sms.models import TrackCreditSMSResource
from passerelle.utils.jsonresponse import APIError
class SendEthicSMSGateway(TrackCreditSMSResource):
account_id = models.CharField(verbose_name=_('Account ID'), max_length=255)
api_key = models.CharField(verbose_name=_('API Key'), max_length=255)
# unecessary field
allow_premium_rate = None
class Meta:
verbose_name = 'Sendethic'
db_table = 'sms_send_ethic'
TEST_DEFAULTS = {
'create_kwargs': {
'account_id': 'yyy',
'api_key': 'www',
'credit_threshold_alert': 1000,
},
'test_vectors': [
{
'status_code': 400,
'response': {'Message': 'Grève des PTT.'},
'result': {
'err': 1,
'err_desc': 'Sendethic error: some destinations failed',
'data': [],
},
},
{
'status_code': 400,
'response': {'no_error_field_in_message': True},
'result': {
'err': 1,
'err_desc': 'Sendethic error: some destinations failed',
'data': [],
},
},
{
'status_code': 200,
'response': '"OK"',
'result': {
'err': 0,
'data': {},
},
},
],
}
URL = 'https://services.message-business.com/api/rest/v4/'
TEST_CREDIT_LEFT = {
'create_kwargs': {
'account_id': 'yyy',
'api_key': 'www',
},
'url': urllib.parse.urljoin(URL, 'account/remainingcredits/sms'),
'get_credit_left_payload': lambda x: {'used': 1, 'remaining': x, 'total': 2000},
}
def request(self, method, endpoint, **kwargs):
url = urllib.parse.urljoin(self.URL, endpoint)
headers = {'Accept': 'application/json'}
try:
response = self.requests.request(
method,
url,
headers=headers,
auth=(self.account_id, self.api_key),
**kwargs,
)
except requests.RequestException as e:
return False, 'Request failed, %s' % e
try:
result = response.json()
if not response.ok:
result = result['Message']
except (ValueError, KeyError):
return False, 'Bad JSON response'
return response.ok, result
def send_msg(self, text, sender, destinations, **kwargs):
# Destination numbers received here were normalized through
# self.clean_number() : they are of the form 00612345678. Remove
# the trailing 00 and replace it with '+', so Sendethic accepts them.
destinations = [f'+{dest[2:]}' for dest in destinations]
errors = []
for dest in destinations:
params = {
'message': text,
'mobile': dest,
}
if sender:
params['oadc'] = sender
success, message = self.request('post', 'sms/send/', json=params)
if not success:
errors.append(message)
if errors:
raise APIError('Sendethic error: some destinations failed', data=errors)
def update_credit_left(self):
success, result = self.request('get', endpoint='account/remainingcredits/sms/')
if not success:
self.logger.warning('Cannot retrieve credits for Sendethic connector: %s', result)
try:
self.credit_left = int(result['remaining'])
self.save(update_fields=['credit_left'])
except KeyError:
self.logger.warning('Cannot retrieve credits for Sendethic connector: %s', result)

61
passerelle/apps/smsfactor/models.py
View File

@ -13,36 +13,19 @@
#
# 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.sms.models import TrackCreditSMSResource
from passerelle.utils.jsonresponse import APIError
class SMSFactorSMSGateway(SMSResource):
class SMSFactorSMSGateway(TrackCreditSMSResource):
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
@ -109,6 +92,14 @@ class SMSFactorSMSGateway(SMSResource):
}
URL = 'https://api.smsfactor.com'
TEST_CREDIT_LEFT = {
'create_kwargs': {
'auth_token': 'yyy',
},
'url': urllib.parse.urljoin(URL, 'credits'),
'get_credit_left_payload': lambda x: {'credits': str(x)},
}
def request(self, method, endpoint, **kwargs):
url = urllib.parse.urljoin(self.URL, endpoint)
@ -173,36 +164,6 @@ class SMSFactorSMSGateway(SMSResource):
try:
# SMS Factor returns this as a string, for an unknown reason
self.credit_left = int(result['credits'])
self.save(update_fields=['credit_left'])
except KeyError:
self.logger.warning('Cannot retrieve credits for sms-factor connector: %s', result)
else:
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 check_status(self):
self.update_credit_left()
self.send_credit_alert_if_needed()

6
passerelle/apps/smsfactor/templates/smsfactor/credit_alert_subject.txt
View File

@ -1,6 +0,0 @@
{% 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 %}

1
passerelle/settings.py
View File

@ -186,6 +186,7 @@ INSTALLED_APPS = (
'passerelle.apps.solis',
'passerelle.apps.twilio',
'passerelle.apps.vivaticket',
'passerelle.apps.sendethic',
# backoffice templates and static
'gadjo',
)

55
passerelle/sms/models.py
View File

@ -13,13 +13,19 @@
#
# 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 re
import urllib.parse
from django.conf import settings
from django.contrib.postgres.fields import ArrayField
from django.core.mail import send_mail
from django.core.validators import RegexValidator
from django.db import models
from django.template.loader import render_to_string
from django.urls import re_path, reverse
from django.utils import timezone
from django.utils.module_loading import import_string
from django.utils.translation import gettext_lazy as _
@ -256,3 +262,52 @@ class SMSLog(models.Model):
def __str__(self):
return '%s %s %s' % (self.timestamp, self.appname, self.slug)
class TrackCreditSMSResource(SMSResource):
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)
def update_credit_left(self):
raise NotImplementedError()
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('sms/credit_alert_subject.txt', ctx).strip()
body = render_to_string('sms/credit_alert_body.txt', ctx)
html_body = render_to_string('sms/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 check_status(self):
self.update_credit_left()
self.send_credit_alert_if_needed()
class Meta:
abstract = True

3
passerelle/apps/smsfactor/templates/smsfactor/credit_alert_body.html → passerelle/sms/templates/sms/credit_alert_body.html
View File

@ -12,7 +12,7 @@
<p>
{% blocktrans trimmed with account=connector.account %}
Please add more credit as soon as possible for your SMS Factor account.
Please add more credit as soon as possible for your account.
{% endblocktrans %}
</p>
@ -20,3 +20,4 @@
{% include "emails/button-link.html" with url=connector_url label=button_label %}
{% endwith %}
{% endblock %}

5
passerelle/apps/smsfactor/templates/smsfactor/credit_alert_body.txt → passerelle/sms/templates/sms/credit_alert_body.txt
View File

@ -7,11 +7,12 @@
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.
{% blocktrans trimmed %}
Please add more credit as soon as possible for your account.
{% endblocktrans %}
{% trans "View connector page:" %} {{ connector_url }}
{% endautoescape %}
{% endblock %}

2
passerelle/apps/ovh/templates/ovh/credit_alert_subject.txt → passerelle/sms/templates/sms/credit_alert_subject.txt
View File

@ -2,5 +2,5 @@
{% load i18n %}
{% block email-subject %}{% autoescape off %}{% blocktrans trimmed with credit_left=connector.credit_left %}
OVH SMS alert: only {{ credit_left }} credits left
SMS alert: only {{ credit_left }} credits left
{% endblocktrans %}{% endautoescape %}{% endblock %}

124
tests/test_sms.py
View File

@ -28,14 +28,18 @@ from passerelle.apps.ovh.models import OVHSMSGateway
from passerelle.apps.sfr_dmc.models import SfrDmcGateway
from passerelle.apps.smsfactor.models import SMSFactorSMSGateway
from passerelle.base.models import AccessRight, ApiUser, Job, ResourceLog
from passerelle.sms.models import SMSLog, SMSResource
from passerelle.sms.models import SMSLog, SMSResource, TrackCreditSMSResource
from passerelle.utils.jsonresponse import APIError
from tests.test_manager import login
from tests.utils import FakedResponse
pytestmark = pytest.mark.django_db
klasses = SMSResource.__subclasses__()
klasses = [
klass
for klass in SMSResource.__subclasses__() + TrackCreditSMSResource.__subclasses__()
if not klass._meta.abstract
]
def test_clean_numbers():
@ -493,60 +497,6 @@ def test_ovh_new_api_credit(app, freezer, admin_user):
connector.check_status()
def test_ovh_alert_emails(app, freezer, mailoutbox):
connector = OVHSMSGateway.objects.create(
slug='test-ovh',
title='Test OVH',
account='sms-test42',
application_key='RHrTdU2oTsrVC0pu',
application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
consumer_key='iF0zi0MJrbjNcI3hvuvwkhNk8skrigxz',
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 = {'creditsLeft': 101}
ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}
with tests.utils.mock_url(ovh_url, resp, 200):
connector.check_status()
assert len(mailoutbox) == 0
resp = {'creditsLeft': 99}
ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}
with tests.utils.mock_url(ovh_url, resp, 200):
connector.check_status()
assert len(mailoutbox) == 1
mail = mailoutbox[0]
assert mail.recipients() == ['test@entrouvert.org']
assert mail.subject == 'OVH SMS alert: only 99 credits left'
for body in (mail.body, mail.alternatives[0][0]):
assert connector.account in body
assert connector.title in body
assert 'http://localhost/ovh/test-ovh/' in body
mailoutbox.clear()
# alert is sent again daily
freezer.move_to('2019-01-01 12:00:00')
resp = {'creditsLeft': 99}
ovh_url = connector.API_URL % {'serviceName': 'sms-test42'}
with tests.utils.mock_url(ovh_url, resp, 200):
connector.check_status()
assert len(mailoutbox) == 0
freezer.move_to('2019-01-02 01:00:07')
with tests.utils.mock_url(ovh_url, resp, 200):
connector.check_status()
assert len(mailoutbox) == 1
def test_ovh_token_request(admin_user, app):
connector = OVHSMSGateway.objects.create(
slug='test-ovh',
@ -780,52 +730,64 @@ def test_sfr_unicode_message(connector):
_check_media_type('unicode message 😀', 'SMSUnicodeLong')
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'],
credit_klasses = TrackCreditSMSResource.__subclasses__()
@pytest.fixture(params=credit_klasses)
def track_credit_connector(request, db):
klass = request.param
kwargs = getattr(klass, 'TEST_CREDIT_LEFT', {}).get('create_kwargs', {})
kwargs.update(
{
'title': klass.__name__,
'slug': klass.__name__.lower(),
'description': klass.__name__,
'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)
c = klass.objects.create(**kwargs)
api = ApiUser.objects.create(username='apiuser', fullname='Api User', description='api')
obj_type = ContentType.objects.get_for_model(c)
# no access check
AccessRight.objects.create(
codename='can_send_messages', apiuser=api, resource_type=obj_type, resource_pk=connector.pk
codename='can_send_messages', apiuser=api, resource_type=obj_type, resource_pk=c.pk
)
return c
def test_send_alert_emails(track_credit_connector, app, freezer, mailoutbox):
get_credit_left_payload = track_credit_connector.TEST_CREDIT_LEFT['get_credit_left_payload']
url = track_credit_connector.TEST_CREDIT_LEFT['url']
freezer.move_to('2019-01-01 00:00:00')
resp = {'credits': '101'}
url = connector.URL
resp = get_credit_left_payload(101)
with tests.utils.mock_url(url, resp, 200):
connector.check_status()
track_credit_connector.check_status()
assert len(mailoutbox) == 0
resp = {'credits': '99'}
url = connector.URL
resp = get_credit_left_payload(99)
with tests.utils.mock_url(url, resp, 200):
connector.check_status()
track_credit_connector.check_status()
assert len(mailoutbox) == 1
mail = mailoutbox[0]
assert mail.recipients() == ['test@entrouvert.org']
assert mail.subject == 'SMS Factor alert: only 99 credits left'
assert mail.subject == 'SMS 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
assert track_credit_connector.title in body
assert track_credit_connector.get_absolute_url() in body
mailoutbox.clear()
# alert is sent again daily
freezer.move_to('2019-01-01 12:00:00')
resp = {'credits': 99}
url = connector.URL
resp = get_credit_left_payload(99)
with tests.utils.mock_url(url, resp, 200):
connector.check_status()
track_credit_connector.check_status()
assert len(mailoutbox) == 0
freezer.move_to('2019-01-02 01:00:07')
with tests.utils.mock_url(url, resp, 200):
connector.check_status()
track_credit_connector.check_status()
assert len(mailoutbox) == 1