Produits Entr'ouvert » Passerelle

            assert isinstance(data['from'], six.text_type), 'from is not a string'

            assert all(map(lambda x: isinstance(x, six.text_type), data['to'])), \

                'to is not a list of strings'

        except (ValueError, AssertionError) as e:

            raise APIError('Payload error: %s' % e)

        data['message'] = data['message'][:self.max_message_length]

        logging.info('sending SMS to %r from %r', data['to'], data['from'])

        stop = not bool('nostop' in request.GET)

        SMSLog.objects.create(appname=self.get_connector_slug(), slug=self.slug)

    class Meta:

        abstract = True

@six.python_2_unicode_compatible

class SMSLog(models.Model):

    timestamp = models.DateTimeField(auto_now_add=True)

import httmock

import pytest

from django.contrib.contenttypes.models import ContentType

from django.utils.encoding import force_text

from passerelle.apps.orange.models import OrangeSMSGateway, OrangeError

from passerelle.utils.jsonresponse import APIError

NETLOC = 'contact-everyone.orange-business.com'

JSON_HEADERS = {'content-type': 'application/json'}

PAYLOAD = {

    'message': 'hello',

    'from': '+33699999999',

    with pytest.raises(OrangeError, match='Orange returned Invalid JSON content'):

        with httmock.HTTMock(mocked_response):

            orange.diffusion('my_token', 'gid2', PAYLOAD['to'], PAYLOAD['message'])

def test_send_msg(app, connector):

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

    assert not resp.json['err']

    # not 201

    with httmock.HTTMock(response_token_ok, response_group_ok, response_500):

import mock

import pytest

from django.contrib.contenttypes.models import ContentType

from passerelle.apps.ovh.models import OVHSMSGateway

from test_manager import login, admin_user

import utils

pytestmark = pytest.mark.django_db

klasses = SMSResource.__subclasses__()

    # no access check

    AccessRight.objects.create(codename='can_send_messages',

                               apiuser=api,

                               resource_type=obj_type,

                               resource_pk=c.pk)

    return c

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

    result = app.post_json(path, params={})

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

    assert result.json['err_desc'].startswith('Payload error: ')

    payload = {

        'message': 'hello',

        'from': '+33699999999',

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

    }

        with utils.mock_url(

                connector.URL,

                test_vector.get('response', ''),

                test_vector.get('status_code', 200)):

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 not 'accessright/add' in resp.text

    app = login(app)

    payload = {

        'message': message_above_limit,

        'from': '+33699999999',

        'to': ['+33688888888'],

    }

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

        send_function.return_value = {}

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

@pytest.mark.parametrize('connector', [OVHSMSGateway], 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(OVHSMSGateway, 'send_msg') as send_function:

        send_function.return_value = {}

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

        assert SMSLog.objects.filter(appname=connector.get_connector_slug(), slug=connector.slug).exists()

class FakedResponse(mock.Mock):

    headers = {}

    def json(self):

        return json_loads(self.content)

    urlmatch_kwargs = {}

    if url:

        parsed = urlparse.urlparse(url)

        if parsed.netloc:

            urlmatch_kwargs['netloc'] = parsed.netloc

        if parsed.path:

            urlmatch_kwargs['path'] = parsed.path

    if not isinstance(response, str):

        response = json.dumps(response)

    @httmock.urlmatch(**urlmatch_kwargs)

    def mocked(url, request):

        return httmock.response(status_code, response, headers, request=request)

    return httmock.HTTMock(mocked)

def make_resource(model_class, **kwargs):

    resource = model_class.objects.create(**kwargs)

    setup_access_rights(resource)

    return resource