Produits Entr'ouvert » Passerelle

# -*- coding: utf-8 -*-

# Generated by Django 1.11.18 on 2021-02-24 10:23

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0027_transaction_id'),

    ]

    operations = [

        migrations.CreateModel(

            name='Esirius',

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

                (

                    'secret_id',

                    models.CharField(blank=True, max_length=128, verbose_name='Application identifier'),

                ),

                ('secret_key', models.CharField(blank=True, max_length=128, verbose_name='Secret Key')),

                (

                    'base_url',

                    models.CharField(

                        help_text='example: http://HOST/ePlanning/webservices/api/',

                        max_length=256,

                        verbose_name='Service URL',

                    ),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True, related_name='_esirius_users_+', related_query_name='+', to='base.ApiUser'

                    ),

                ),

            ],

            options={

                'verbose_name': 'e-Sirius',

            },

        ),

    ]

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

from datetime import datetime

from urllib.parse import urljoin

from Cryptodome.Cipher import DES

from Cryptodome.Util.Padding import pad

from django.db import models

from django.utils.encoding import force_bytes

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

CREATE_APPOINTMENT_SCHEMA = {

    '$schema': 'http://json-schema.org/draft-04/schema#',

    "type": "object",

    'properties': {

        'idSys': {'type': 'integer'},

        'CodeRDV': {'type': 'string'},

        'beginDate': {'type': 'string', 'pattern': '^[0-9]{4}-[0-9]{2}-[0-9]{2}$'},

        'beginTime': {'type': 'string', 'pattern': '^[0-9]{2}:[0-9]{2}$'},

        'endDate': {'type': 'string', 'pattern': '^[0-9]{4}-[0-9]{2}-[0-9]{2}$'},

        'endTime': {'type': 'string', 'pattern': '^[0-9]{2}:[0-9]{2}$'},

        'comment': {'type': 'string'},

        'isoLanguage': {'description': 'ex: fr', 'type': 'string'},

        'needsConfirmation': {'type': 'boolean'},

        'rdvChannel': {'description': 'ex: EAPP0', 'type': 'string'},

        'receptionChannel': {'type': 'string'},

        'owner': {'type': 'object', 'properties': {'key': {'type': 'string'}, 'value': {'type': 'string'}}},

        'user': {

            'type': 'object',

            'properties': {

                'idSys': {'type': 'integer'},

                'personalIdentity': {'type': 'string'},

                'additionalPersonalIdentity': {"type": "array", "items": {'type': 'string'}},

                'lastName': {'type': 'string'},

                'civility': {'type': 'string'},

                'firstName': {'type': 'string'},

                'birthday': {'type': 'string'},

                'email': {'type': 'string'},

                'fixPhone': {'type': 'string'},

                'phone': {'type': 'string'},

                'address': {

                    'type': 'object',

                    'properties': {

                        'line1': {'type': 'string'},

                        'line2': {'type': 'string'},

                        'zipCode': {'type': 'string'},

                        'city': {'type': 'string'},

                        'country': {'type': 'string'},

                    },

                },

            },

        },

        'serviceId': {'type': 'string'},

        'siteCode': {'type': 'string'},

        "resources": {

            'type': 'object',

            'properties': {

                'id': {'type': 'integer'},

                'key': {'type': 'string'},

                'type': {'type': 'string'},

                'name': {'type': 'string'},

                'station': {

                    'type': 'object',

                    'properties': {

                        'id': {'type': 'integer'},

                        'key': {'type': 'string'},

                        'name': {'type': 'string'},

                    },

                },

            },

        },

        'motives': {

            "type": "array",

            "items": {

                'type': 'object',

                'properties': {

                    'id': {'type': 'integer'},

                    'name': {'type': 'string'},

                    'shortName': {'type': 'string'},

                    'processingTime': {'type': 'integer'},

                    'externalModuleAccess': {'type': 'integer'},

                    'quantity': {'type': 'integer'},

                    'usePremotiveQuantity': {'type': 'boolean'},

                },

            },

        },

    },

    'unflatten': True,

}

class ESirius(BaseResource):

    secret_id = models.CharField(max_length=128, verbose_name=_('Application identifier'), blank=True)

    secret_key = models.CharField(max_length=128, verbose_name=_('Secret Key'), blank=True)

    base_url = models.CharField(

        max_length=256,

        blank=False,

        verbose_name=_('Service URL'),

        help_text=_('example: http://HOST/ePlanning/webservices/api/'),

    )

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('e-Sirius')

    @staticmethod

    def tokenize(caller, key, epoch):

        des_key = pad(force_bytes(key), 8)[:8]

        cipher = DES.new(des_key, DES.MODE_ECB)

        # error 500 if spaces are inserted

        plaintext = '{"caller":"%s","createInfo":%i}' % (caller, epoch)

        msg = cipher.encrypt(pad(force_bytes(plaintext), 8))

        return base64.b64encode(msg)

    def pre_request(self, uri):

        url = urljoin(self.base_url, uri)

        epoch = int(datetime.now().strftime('%s') + '000')

        headers = {

            'Accept': 'application/json; charset=utf-8',

            'token_info_caller': ESirius.tokenize(self.secret_id, self.secret_key, epoch),

        }

        return url, headers

    @staticmethod

    def post_request(response):

        if response.status_code != 200:

            try:

                json_content = response.json()

            except ValueError:

                json_content = None

            raise APIError(

                'error status:%s %r, content:%r'

                % (response.status_code, response.reason, response.text[:1024]),

                data={'status_code': response.status_code, 'json_content': json_content},

            )

    def check_status(self):

        """

        Raise an exception if something goes wrong.

        """

        url, headers = self.pre_request('sites/')

        response = self.requests.get(url, headers=headers)

        ESirius.post_request(response)

    @endpoint(

        display_category=_('Appointment'),

        name='create-appointment',

        perm='can_access',

        methods=['post'],

        description=_('Create appointment'),

        post={'request_body': {'schema': {'application/json': CREATE_APPOINTMENT_SCHEMA}}},

    )

    def create_appointment(self, request, post_data):

        # address dict is required

        if not post_data.get('user'):

            post_data.update({'user': {}})

        if not post_data['user'].get('address'):

            post_data['user'].update({'address': {}})

        url, headers = self.pre_request('appointments/')

        response = self.requests.post(url, json=post_data, headers=headers)

        ESirius.post_request(response)

        return {'data': {'booking_id': response.text}}

    @endpoint(

        display_category=_('Appointment'),

        name='delete-appointment',

        perm='can_access',

        methods=['post'],

        description=_('Delete appointment'),

        parameters={

            'booking_id': {'description': _('id returned by create-appointment endpoint')},

        },

    )

    def delete_appointment(self, request, booking_id):

        url, headers = self.pre_request('appointments/%s/' % booking_id)

        response = self.requests.delete(url, headers=headers)

        if response.status_code == 304:

            raise APIError('Booking not found: %s' % booking_id)

        ESirius.post_request(response)

        return {'data': {}}

    'passerelle.apps.esirius',

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2021  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 httmock

import pytest

from passerelle.apps.esirius.models import ESirius

from passerelle.utils.jsonresponse import APIError

import utils

CREATE_APPOINTMENT_PAYLOAD = {

    'beginDate': '2021-02-24',

    'beginTime': '16:40',

    'endDate': '2021-02-24',

    'endTime': '17:00',

    'comment': 'commentaire',

    'isoLanguage': 'fr',

    'needsConfirmation': False,

    'rdvChannel': 'WEBSERVICES',

    'receptionChannel': 'WS',

    'serviceId': '9',

    'siteCode': 'site1',

    'resources': {

        'id': 1,

        'key': '17',

        'type': 'STATION',

    },

}

@pytest.fixture

def connector(db):

    return utils.setup_access_rights(

        ESirius.objects.create(

            slug='test', secret_id='xxx', secret_key='yyy', base_url='https://dummy-server.org'

        )

    )

def get_endpoint(name):

    return utils.generic_endpoint_url('esirius', name)

def test_tokenise(connector):

    assert (

        ESirius.tokenize('eAppointment', 'ES2I Info Caller Http Encryption Key', 1611673986880)

        == b'yM4zYAxT67Qvjd20riG3j0eu0t0Ku+HLlttj17Gul7zkruFaXX1J8BJ6sV2Ldgw40axfWh+ESAY='

    )

def test_pre_request(connector):

    url, headers = connector.pre_request('an/uri/')

    assert url == 'https://dummy-server.org/an/uri/'

    assert headers['Accept'] == 'application/json; charset=utf-8'

    assert headers['token_info_caller'][:42] == b'f3G6sjRZETBam6vcdrAxmvJQTX5hh6OjZ8XlUO6SMo'

    assert headers['token_info_caller'][-10:] == b'DbrIGCxaCW'

@pytest.mark.parametrize(

    'status_code, content, a_dict',

    [

        (400, '{"message": "help"}', {'message': 'help'}),

        (500, 'not json', None),

    ],

)

def test_post_request(status_code, content, a_dict):

    with pytest.raises(APIError) as exc:

        ESirius.post_request(httmock.response(status_code, content))

    assert exc.value.err

    assert exc.value.data['status_code'] == status_code

    assert exc.value.data['json_content'] == a_dict

@pytest.mark.parametrize(

    'status_code, content, is_up',

    [

        (200, 'wathever', True),

        (500, '{"message": "help"}', False),

    ],

)

def test_check_status(app, connector, status_code, content, is_up):

    @httmock.all_requests

    def sigerly_mock(url, request):

        return httmock.response(status_code, content)

    if is_up:

        with httmock.HTTMock(sigerly_mock):

            connector.check_status()

    else:

        with pytest.raises(APIError):

            with httmock.HTTMock(sigerly_mock):

                connector.check_status()

def test_create_appointment(app, connector):

    endpoint = get_endpoint('create-appointment')

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/', method='POST')

    def sigerly_mock(url, request):

        return httmock.response(200, b'94PEP4')

    with httmock.HTTMock(sigerly_mock):

        resp = app.post_json(endpoint, params=CREATE_APPOINTMENT_PAYLOAD)

    assert not resp.json['err']

    assert resp.json['data'] == {'booking_id': '94PEP4'}

def test_create_appointment_error_404(app, connector):

    endpoint = get_endpoint('create-appointment')

    # payload not providing or probiding an unconfigured serviceId

    payload = CREATE_APPOINTMENT_PAYLOAD

    del payload['serviceId']

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/', method='POST')

    def sigerly_mock(url, request):

        return httmock.response(

            404,

            {

                'code': 'Not Found',

                'type': 'com.es2i.planning.api.exception.NoService4RDVException',

                'message': "Le rendez-vous {0} n'a pas créé",

            },

        )

    with httmock.HTTMock(sigerly_mock):

        resp = app.post_json(endpoint, params=payload)

    assert resp.json['err']

    assert resp.json['data']['status_code'] == 404

    assert resp.json['data']['json_content'] == {

        'code': 'Not Found',

        'type': 'com.es2i.planning.api.exception.NoService4RDVException',

        'message': "Le rendez-vous {0} n'a pas créé",

    }

def test_create_appointment_error_500(app, connector):

    endpoint = get_endpoint('create-appointment')

    # payload not providing beginTime

    payload = {'beginDate': '2021-02-23'}

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/', method='POST')

    def sigerly_mock(url, request):

        return httmock.response(500, 'java stack')

    with httmock.HTTMock(sigerly_mock):

        resp = app.post_json(endpoint, params=payload)

    assert resp.json['err']

    assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'

    assert resp.json['err_desc'] == "error status:500 None, content:'java stack'"

    assert resp.json['data']['status_code'] == 500

    assert resp.json['data']['json_content'] is None

def test_delete_appointment(app, connector):

    endpoint = get_endpoint('delete-appointment')

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/', method='DELETE')

    def sigerly_mock(url, request):

        return httmock.response(200, b'')

    with httmock.HTTMock(sigerly_mock):

        resp = app.post_json(endpoint + '?booking_id=94PEP4')

    assert not resp.json['err']

    assert resp.json['data'] == {}

def test_delete_appointment_error(app, connector):

    endpoint = get_endpoint('delete-appointment')

    @httmock.urlmatch(netloc='dummy-server.org', path='/appointments/', method='DELETE')

    def sigerly_mock(url, request):

        return httmock.response(304, b'')

    with httmock.HTTMock(sigerly_mock):

        resp = app.post_json(endpoint + '?booking_id=94PEP4')

    assert resp.json['err']

    assert resp.json['err_desc'] == 'Booking not found: 94PEP4'