# 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 logging
from unittest import mock

import pytest
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.utils.translation import gettext as _

import tests.utils
from passerelle.apps.choosit.models import ChoositSMSGateway
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, TrackCreditSMSResource
from passerelle.utils.jsonresponse import APIError
from tests.test_manager import login
from tests.utils import FakedResponse

pytestmark = pytest.mark.django_db

klasses = [
    klass
    for klass in SMSResource.__subclasses__() + TrackCreditSMSResource.__subclasses__()
    if not klass._meta.abstract
]


def test_clean_numbers():
    connector = OVHSMSGateway()
    assert connector.clean_numbers(['+ 33 12']) == ['003312']
    assert connector.clean_numbers(['0 0 33 12']) == ['003312']
    assert connector.clean_numbers(['0 12']) == ['003312']
    connector.default_country_code = '32'
    connector.default_trunk_prefix = '1'
    connector.save()
    assert connector.clean_numbers(['+ 33 12']) == ['003312']
    assert connector.clean_numbers(['0 0 33 12']) == ['003312']
    assert connector.clean_numbers(['1 12']) == ['003212']
    with pytest.raises(APIError, match='phone number %r is unsupported' % '0123'):
        connector.clean_numbers(['0123'])


def test_authorize_numbers():
    connector = OVHSMSGateway()

    # premium-rate
    assert connector.allow_premium_rate is False
    number = '0033' + '8' + '12345678'
    with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
        connector.authorize_numbers([number])
    connector.allow_premium_rate = True
    connector.save()
    assert connector.authorize_numbers([number])[0] == [number]

    # All country
    assert connector.authorized == [SMSResource.ALL]
    number = '0033' + '1' + '12345678'
    assert connector.authorize_numbers([number])[0] == [number]
    connector.authorized = [SMSResource.FR_METRO]
    connector.save()
    with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
        connector.authorize_numbers([number])

    # France
    number = '0033' + '6' + '12345678'
    assert connector.authorize_numbers([number])[0] == [number]
    connector.authorized = [SMSResource.FR_DOMTOM]
    connector.save()
    with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
        connector.authorize_numbers([number])

    # Dom-Tom
    number = '596596' + '123456'
    assert connector.authorize_numbers([number])[0] == [number]
    connector.authorized = [SMSResource.BE_]
    connector.save()
    with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
        connector.authorize_numbers([number])

    # Belgian
    number = '0032' + '45' + '1234567'
    assert connector.authorize_numbers([number])[0] == [number]
    connector.authorized = [SMSResource.FR_METRO]
    connector.save()
    with pytest.raises(APIError, match='no phone number was authorized: %s' % number):
        connector.authorize_numbers([number])

    # Don't raise if authorized destinations are not empty
    connector.allow_premium_rate = False
    connector.authorized = [SMSResource.FR_METRO]
    connector.save()
    numbers = [
        '0033' + '8' + '12345678',
        '0033' + '1' + '12345678',
        '0033' + '6' + '12345678',
        '596596' + '123456',
        '0032' + '45' + '1234567',
    ]
    authorized_numbers, warnings = connector.authorize_numbers(numbers)
    assert authorized_numbers == ['0033612345678']
    assert warnings == {
        'deny premium rate phone numbers': '0033812345678',
        'deny foreign phone numbers': '0032451234567, 0033112345678, 596596123456',
    }


@pytest.fixture(params=klasses)
def connector(request, db):
    klass = request.param
    kwargs = getattr(klass, 'TEST_DEFAULTS', {}).get('create_kwargs', {})
    kwargs.update(
        {
            'title': klass.__name__,
            'slug': klass.__name__.lower(),
            'description': klass.__name__,
        }
    )
    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=c.pk
    )
    return c


def test_connectors(app, connector, freezer):
    path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
    result = app.post_json(path, params={}, status=400)
    assert result.json['err'] == 1
    assert result.json['err_desc'] == "'message' is a required property"

    payload = {
        'message': 'hello',
        'from': '+33699999999',
        'to': ['+33688888888', '+33677777777'],
    }
    test_vectors = getattr(connector, 'TEST_DEFAULTS', {}).get('test_vectors', [])
    total = len(test_vectors)
    nb_failed = 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(method_name='send_job', status='registered').id

        # perform job
        freezer.move_to('2019-01-01 01:00:03')
        with tests.utils.mock_url(
            connector.URL, test_vector.get('response', ''), test_vector.get('status_code', 200)
        ):
            connector.jobs()
        job = Job.objects.get(id=job_id)
        if job.status == 'failed':
            assert len(job.status_details['error_summary']) > 0
            assert test_vector['result']['err_desc'] in job.status_details['error_summary']
            nb_failed += 1
        else:
            assert job.status == 'completed'
    assert Job.objects.filter(method_name='send_job').count() == total
    assert SMSLog.objects.count() == total - nb_failed


@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
def test_sms_legacy_retry_after_error(app, connector, freezer):
    path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
    assert not SMSLog.objects.filter(appname=connector.get_connector_slug(), slug=connector.slug).exists()

    payload = {
        'message': 'plop',
        'from': '+33699999999',
        'to': ['+33688888888'],
    }
    freezer.move_to('2019-01-01 00:00:00')
    result = app.post_json(path, params=payload)
    assert result.json['err'] == 0
    assert Job.objects.exists()
    assert Job.objects.get().status == 'registered'
    assert not Job.objects.get().after_timestamp
    with tests.utils.mock_url(
        url=connector.URL,
        response={'status': 429, 'message': 'Too much requests.\nPlease retry in 3 seconds.'},
    ):
        connector.jobs()
    assert Job.objects.get().status == 'registered'
    assert Job.objects.get().after_timestamp

    with tests.utils.mock_url(url=connector.URL, response={'status': 100, 'credit_left': 22}):
        connector.jobs()
    assert Job.objects.get().status == 'registered'

    freezer.move_to('2019-01-01 00:00:11')
    with tests.utils.mock_url(url=connector.URL, response={'status': 100, 'creditLeft': 22}):
        connector.jobs()
    assert Job.objects.get().status == 'completed'
    assert SMSLog.objects.exists()


def test_manage_views(admin_user, app, connector):
    url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
    resp = app.get(url)
    assert 'Endpoints' in resp.text
    assert 'accessright/add' not in resp.text
    app = login(app)
    resp = app.get(url)
    description_fields = [
        x.text.split(':')[0] for x in resp.html.find('div', {'id': 'description'}).find_all('p')
    ]
    assert 'Default country code' in description_fields
    assert 'Default trunk prefix' in description_fields
    assert 'Maximum message length' in description_fields
    assert 'Account' not in description_fields
    assert 'Username' not in description_fields
    assert 'Endpoints' in resp.text
    assert 'accessright/add' in resp.text


@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
def test_manage_views_ovh(app, connector, admin_user):
    login(app)
    connector.default_country_code = '44'
    connector.account = 'secret'
    connector.application_key = 'secret'
    connector.application_secret = 'secret'
    connector.consumer_key = 'secret'
    connector.password = 'secret'
    connector.username = 'secret'
    connector.alert_emails = ['test@entrouvert.org', 'foo@example.com']
    connector.save()
    url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
    resp = app.get(url)
    description_fields = [x.text for x in resp.html.find('div', {'id': 'description'}).find_all('p')]
    assert any(x for x in description_fields if 'Default country code' in x)
    assert any(x for x in description_fields if '44' in x)
    assert not any(x for x in description_fields if 'secret' in x)
    alert_emails_filed = [x for x in description_fields if 'send credit alerts to' in x]
    assert alert_emails_filed[0].split(':')[1].strip() == 'test@entrouvert.org, foo@example.com'


@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
def test_sms_max_message_length(app, connector):
    path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)

    message_above_limit = 'a' * (connector.max_message_length + 1)
    payload = {
        'message': message_above_limit,
        'from': '+33699999999',
        'to': ['+33688888888'],
    }
    with mock.patch.object(OVHSMSGateway, 'send_msg') as send_function:
        send_function.return_value = None
        app.post_json(path, params=payload)
        connector.jobs()
        assert send_function.call_args[1]['text'] == 'a' * connector.max_message_length


@pytest.mark.parametrize('connector', [OVHSMSGateway, SMSFactorSMSGateway], indirect=True)
def test_sms_log(app, connector):
    path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
    assert not SMSLog.objects.filter(appname=connector.get_connector_slug(), slug=connector.slug).exists()

    payload = {
        'message': 'plop',
        'from': '+33699999999',
        'to': ['+33688888888'],
    }

    with mock.patch.object(connector.__class__, 'send_msg') as send_function:
        send_function.return_value = 1
        app.post_json(path, params=payload)
        connector.jobs()
        assert SMSLog.objects.filter(
            appname=connector.get_connector_slug(), slug=connector.slug, credits=1
        ).exists()

    with mock.patch.object(connector.__class__, 'send_msg') as send_function:
        send_function.return_value = 2
        app.post_json(path, params=payload)
        connector.jobs()
        assert SMSLog.objects.filter(
            appname=connector.get_connector_slug(), slug=connector.slug, credits=2
        ).exists()


@pytest.mark.parametrize('connector', [OVHSMSGateway, SMSFactorSMSGateway], indirect=True)
def test_sms_job_details_credits(admin_user, app, connector, caplog):
    path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
    payload = {'message': 'plop', 'from': '+33699999999', 'to': ['+33688888888']}
    with mock.patch.object(connector.__class__, 'send_msg') as send_function:
        send_function.return_value = 1
        app.post_json(path, params=payload)
        job1_id = Job.objects.get(method_name='send_job', status='registered').id
        connector.jobs()

        send_function.return_value = 2
        app.post_json(path, params=payload)
        job2_id = Job.objects.get(method_name='send_job', status='registered').id
        connector.jobs()

    app = login(app)
    resp = app.get('/manage/%s/%s/jobs/%s/' % (connector.get_connector_slug(), connector.slug, job1_id))
    assert len(resp.pyquery('td:contains("credits-spent")')) == 1
    assert len(resp.pyquery('td:contains("1")')) == 1

    resp = app.get('/manage/%s/%s/jobs/%s/' % (connector.get_connector_slug(), connector.slug, job2_id))
    assert len(resp.pyquery('td:contains("credits-spent")')) == 1
    assert len(resp.pyquery('td:contains("2")')) == 1


def test_sms_nostop_parameter(app, connector):
    base_path = '/%s/%s/send/?nostop=1' % (connector.get_connector_slug(), connector.slug)
    payload = {
        'message': 'not a spam',
        'from': '+33699999999',
        'to': ['+33688888888'],
    }
    for path in (base_path, base_path + '?nostop=1', base_path + '?nostop=foo', base_path + '?nostop'):
        send_patch = mock.patch(
            'passerelle.apps.%s.models.%s.send_msg'
            % (connector.__class__._meta.app_label, connector.__class__.__name__)
        )
        with send_patch as send_function:
            send_function.return_value = None
            app.post_json(base_path, params=payload)
            connector.jobs()
            assert send_function.call_args[1]['text'] == 'not a spam'
            assert send_function.call_args[1]['stop'] == ('nostop' not in path)


@pytest.mark.parametrize('connector', [OVHSMSGateway, SMSFactorSMSGateway], indirect=True)
@pytest.mark.parametrize(
    'to, destination',
    [
        ('06 12 34 56 78', '0033612345678'),
        ('06.12.34.56.78', '0033612345678'),
        ('06-12-34-56-78', '0033612345678'),
        ('+33/612345678', '0033612345678'),
    ],
)
def test_send_schema(app, connector, to, destination):
    base_path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
    payload = {
        'message': 'not a spam',
        'from': '+33699999999',
        'to': [to],
    }
    send_patch = mock.patch(
        'passerelle.apps.%s.models.%s.send_msg'
        % (connector.__class__._meta.app_label, connector.__class__.__name__)
    )
    with send_patch as send_function:
        send_function.return_value = None
        app.post_json(base_path, params=payload)
        connector.jobs()
        assert send_function.call_args[1]['destinations'] == [destination]


def test_ovh_new_api(app, freezer):
    connector = OVHSMSGateway.objects.create(
        slug='ovh',
        account='sms-test42',
        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(method_name='send_job', status='registered').id

    # perform job
    freezer.move_to('2019-01-01 01:00:03')
    resp = {
        'validReceivers': ['+33688888888', '+33677777777'],
        'totalCreditsRemoved': 1,
        'ids': [241615100],
        'invalidReceivers': [],
    }
    base_url = connector.API_URL % {'serviceName': 'sms-test42'}
    url = base_url + 'jobs/'
    with tests.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


@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
def test_sms_test_send(admin_user, app, connector):
    url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
    resp = app.get(url)
    assert 'Send a test message' not in resp.text

    app = login(app)
    resp = app.get(url)
    assert 'Send a test message' in resp.text
    assert resp.pyquery('a#sms-test-send')[0].attrib['href'] == reverse(
        'sms-test-send', kwargs={'connector': connector.get_connector_slug(), 'slug': connector.slug}
    )

    resp = resp.click('Send a test message')
    resp.form['number'] = '+33688888888'
    resp.form['sender'] = '+33699999999'
    resp.form['message'] = 'hello'
    with mock.patch.object(OVHSMSGateway, 'send_msg') as send_function:
        send_function.return_value = None
        resp = resp.form.submit()
        assert send_function.call_args[1] == {
            'text': 'hello',
            'sender': '+33699999999',
            'destinations': ['0033688888888'],
            'stop': False,
        }
    assert resp.status_code == 302
    assert resp.location == url


def test_ovh_new_api_credit(app, freezer, admin_user):
    login(app)
    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 tests.utils.mock_url(ovh_url, resp, 200):
        connector.jobs()
    connector.refresh_from_db()
    assert connector.credit_left == 123

    resp = app.get(manager_url)
    assert '123' in resp.text

    # credit update when checking status
    resp = {
        'creditsLeft': 456,
    }
    with tests.utils.mock_url(ovh_url, resp, 200):
        connector.check_status()
    assert connector.credit_left == 456

    resp = {
        'message': 'This credential is not valid',
        'httpCode': '403 Forbidden',
        'errorCode': 'INVALID_CREDENTIAL',
    }
    with tests.utils.mock_url(ovh_url, resp, 403):
        with pytest.raises(APIError, match='Forbidden'):
            connector.check_status()


def test_ovh_token_request(admin_user, app):
    connector = OVHSMSGateway.objects.create(
        slug='test-ovh',
        title='Test OVH',
        account='sms-test42',
        application_key='RHrTdU2oTsrVC0pu',
        application_secret='CLjtS69tTcPgCKxedeoZlgMSoQGSiXMa',
    )

    app = login(app)
    resp = app.get(connector.get_absolute_url())
    assert 'not operational yet' in resp.text

    ovh_request_token_url = 'https://eu.api.ovh.com/1.0/auth/credential'
    ovh_response = {
        'consumerKey': 'xyz',
        'validationUrl': 'https://eu.api.ovh.com/auth/?credentialToken=iQ1joJE',
    }
    with tests.utils.mock_url(ovh_request_token_url, ovh_response, 302) as mocked:
        resp = resp.click('request access')
    assert resp.url == 'https://eu.api.ovh.com/auth/?credentialToken=iQ1joJE'

    request = mocked.handlers[0].call['requests'][0]
    body = json.loads(request.body.decode())
    assert 'accessRules' in body
    redirect_url = body['redirection'][len('http://testserver') :]

    resp = app.get(redirect_url).follow()
    assert 'Successfuly completed connector configuration' in resp.text
    connector.refresh_from_db()
    assert connector.consumer_key == 'xyz'


def test_ovh_token_request_error(admin_user, app):
    connector = OVHSMSGateway.objects.create(
        slug='test-ovh',
        title='Test OVH',
        account='sms-test42',
        application_key='wrong',
        application_secret='oups',
    )

    app = login(app)
    resp = app.get(connector.get_absolute_url())
    ovh_request_token_url = 'https://eu.api.ovh.com/1.0/auth/credential'
    ovh_response = {'message': 'Invalid application key'}

    with tests.utils.mock_url(ovh_request_token_url, ovh_response, 401):
        resp = resp.click('request access').follow()
    assert 'error requesting token: Invalid application key.' in resp.text

    ovh_response = 'not-json'
    with tests.utils.mock_url(ovh_request_token_url, ovh_response, 401):
        resp = resp.click('request access').follow()
    assert 'error requesting token: bad JSON response' in resp.text


@pytest.mark.parametrize('connector', [ChoositSMSGateway], indirect=True)
def test_manager(admin_user, app, connector):
    app = login(app)
    path = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
    resp = app.get(path)
    assert (
        '33'
        in [
            x.text
            for x in resp.html.find('div', {'id': 'description'}).find_all('p')
            if x.text.startswith(_('Default country code'))
        ][0]
    )
    assert (
        _('All')
        in [x.text for x in resp.html.find_all('p') if x.text.startswith(_('Authorized Countries'))][0]
    )
    assert (
        _('no')
        in [x.text for x in resp.html.find_all('p') if x.text.startswith(_('Allow premium rate numbers'))][0]
    )

    path = '/manage/%s/%s/edit' % (connector.get_connector_slug(), connector.slug)
    resp = app.get(path)
    resp.form['authorized'] = []
    resp.form['default_country_code'] = '+33'
    resp.form['default_trunk_prefix'] = 'x'
    resp = resp.form.submit()
    assert resp.html.find('div', {'class': 'errornotice'}).p.text == 'There were errors processing your form.'
    assert [x.text.strip() for x in resp.html.find_all('div', {'class': 'error'})] == [
        'The country must only contain numbers',
        'The trunk prefix must only contain numbers',
        'This field is required.',
    ]
    resp.form['authorized'] = [SMSResource.FR_METRO, SMSResource.FR_DOMTOM]
    resp.form['default_country_code'] = '33'
    resp.form['default_trunk_prefix'] = '0'
    resp = resp.form.submit()
    resp = resp.follow()
    assert (
        _('France mainland (+33 [67])')
        in [x.text for x in resp.html.find_all('p') if x.text.startswith(_('Authorized Countries'))][0]
    )

    path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
    payload = {
        'message': 'plop',
        'from': '+33699999999',
        'to': ['+33688888888'],
    }
    resp = app.post_json(path, params=payload)
    assert resp.json['warn'] == {
        'deny premium rate phone numbers': '',
        'deny foreign phone numbers': '',
    }
    with mock.patch.object(type(connector), 'send_msg') as send_function:
        send_function.return_value = None
        connector.jobs()
    assert SMSLog.objects.count() == 1

    payload['to'][0] = '+33188888888'
    SMSLog.objects.all().delete()
    app.post_json(path, params=payload)
    with mock.patch.object(type(connector), 'send_msg') as send_function:
        send_function.return_value = None
        connector.jobs()
    assert not SMSLog.objects.count()
    assert ResourceLog.objects.filter(levelno=logging.WARNING).count() == 1
    assert (
        ResourceLog.objects.filter(levelno=30)[0].extra['exception']
        == 'no phone number was authorized: 0033188888888'
    )


@pytest.mark.parametrize('connector', [OVHSMSGateway], indirect=True)
def test_api_statistics(app, freezer, connector, admin_user):
    resp = app.get('/api/statistics/')
    url = [x for x in resp.json['data'] if x['id'] == 'sms_count_ovh_ovhsmsgateway'][0]['url']

    assert app.get(url, status=403)

    login(app)
    resp = app.get(url)
    assert len(resp.json['data']['series'][0]['data']) == 0

    freezer.move_to('2021-01-01 12:00')
    for i in range(5):
        SMSLog.objects.create(appname='ovh', slug='ovhsmsgateway')

    freezer.move_to('2021-02-03 13:00')
    for i in range(3):
        SMSLog.objects.create(appname='ovh', slug='ovhsmsgateway')

    freezer.move_to('2021-02-06 13:00')
    SMSLog.objects.create(appname='ovh', slug='ovhsmsgateway')
    SMSLog.objects.create(appname='ovh', slug='other')

    resp = app.get(url + '?time_interval=day')
    assert resp.json['data'] == {
        'x_labels': ['2021-01-01', '2021-02-03', '2021-02-06'],
        'series': [{'label': 'SMS Count', 'data': [5, 3, 1]}],
    }

    resp = app.get(url + '?start=2021-02-04&end=2021-02-07')
    assert resp.json['data'] == {
        'x_labels': ['2021-02-06'],
        'series': [{'label': 'SMS Count', 'data': [1]}],
    }

    # invalid time_interval
    resp = app.get(url + '?time_interval=month')
    assert resp.json['err'] == 1


@pytest.mark.parametrize('connector', [SfrDmcGateway], indirect=True)
def test_sfr_max_message_error(connector):
    with pytest.raises(APIError, match='you can\'t send more than 20 sms at once.'):
        connector.send_msg('hello', '0033699999999', [f'00336888888{suffix:02d}' for suffix in range(0, 21)])


@pytest.mark.parametrize('connector', [SfrDmcGateway], indirect=True)
def test_sfr_prefix(connector):
    json_response = {
        'success': True,
        'response': {
            0: '4500283767',
            1: '4500283767',
        },
    }

    with mock.patch('passerelle.utils.Request.get') as mocked_get:
        mocked_get.return_value = FakedResponse(content=json.dumps(json_response), status_code=200)
        connector.send_msg('hello', '0033699999999', ['0033688888888'])
        mocked_get.assert_called_once_with(
            'https://www.dmc.sfr-sh.fr/DmcWS/1.5.7/JsonService/MessagesUnitairesWS/batchSingleCall',
            params={
                'authenticate': json.dumps(
                    {'serviceId': '1234', 'servicePassword': 'krascuky', 'spaceId': '1234'}
                ),
                'messageUnitaires': json.dumps(
                    [
                        {
                            'media': 'SMSLong',
                            'textMsg': 'hello',
                            'from': '0033699999999',
                            'to': '+33688888888',
                        },
                    ]
                ),
            },
        )


@pytest.mark.parametrize('connector', [SfrDmcGateway], indirect=True)
def test_sfr_unicode_message(connector):
    json_response = {
        'success': True,
        'response': {
            0: '4500283767',
            1: '4500283767',
        },
    }

    def _check_media_type(message, expected_media_type):
        with mock.patch('passerelle.utils.Request.get') as mocked_get:
            mocked_get.return_value = FakedResponse(content=json.dumps(json_response), status_code=200)
            connector.send_msg(message, '0033699999999', ['0033688888888'])
            mocked_get.assert_called_once()
            messages = json.loads(mocked_get.call_args[1]['params']['messageUnitaires'])
            assert messages[0]['media'] == expected_media_type

    _check_media_type('standard GSM message', 'SMSLong')
    _check_media_type('usual standard GSM characters : \'"-\r\n!?éèù%à*+=€@[]|', 'SMSLong')
    _check_media_type('unicode message 😀', 'SMSUnicodeLong')


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'],
        }
    )
    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=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 = get_credit_left_payload(101)
    with tests.utils.mock_url(url, resp, 200):
        track_credit_connector.check_status()
    assert len(mailoutbox) == 0

    resp = get_credit_left_payload(99)
    with tests.utils.mock_url(url, resp, 200):
        track_credit_connector.check_status()
    assert len(mailoutbox) == 1

    mail = mailoutbox[0]
    assert mail.recipients() == ['test@entrouvert.org']
    assert mail.subject == 'SMS alert: only 99 credits left'
    for body in (mail.body, mail.alternatives[0][0]):
        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 = get_credit_left_payload(99)
    with tests.utils.mock_url(url, resp, 200):
        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):
        track_credit_connector.check_status()
    assert len(mailoutbox) == 1