Produits Entr'ouvert » Passerelle

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

# Generated by Django 1.11.18 on 2021-02-26 14:14

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

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

                (

                    'basic_auth_username',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication username'

                    ),

                ),

                (

                    'basic_auth_password',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication password'

                    ),

                ),

                (

                    'client_certificate',

                    models.FileField(

                        blank=True, null=True, upload_to='', verbose_name='TLS client certificate'

                    ),

                ),

                (

                    'trusted_certificate_authorities',

                    models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'),

                ),

                ('verify_cert', models.BooleanField(default=True, verbose_name='TLS verify certificates')),

                (

                    'http_proxy',

                    models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'),

                ),

                (

                    '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: https://HOST/ePlanning/webservices/api/',

                        max_length=256,

                        verbose_name='ePlanning webservices URL',

                    ),

                ),

                (

                    'users',

                    models.ManyToManyField(

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

                    ),

                ),

            ],

            options={

                'verbose_name': '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 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, HTTPResource

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': 'string', 'pattern': '^[0-9]*$'},

        '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': {'description': 'boolean expected', 'type': 'string'},

        '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': 'string', 'pattern': '^[0-9]*$'},

                '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': 'string', 'pattern': '^[0-9]*$'},

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

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

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

                'station': {

                    'type': 'object',

                    'properties': {

                        'id': {'type': 'string', 'pattern': '^[0-9]*$'},

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

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

                    },

                },

            },

        },

        'motives': {

            "type": "array",

            "items": {

                'type': 'object',

                'properties': {

                    'id': {'type': 'string', 'pattern': '^[0-9]*$'},

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

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

                    'processingTime': {'type': 'string', 'pattern': '^[0-9]*$'},

                    'externalModuleAccess': {'type': 'string', 'pattern': '^[0-9]*$'},

                    'quantity': {'type': 'string', 'pattern': '^[0-9]*$'},

                    'usePremotiveQuantity': {'description': 'boolean expected', 'type': 'string'},

                },

            },

        },

    },

    'unflatten': True,

}

class ESirius(BaseResource, HTTPResource):

    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=_('ePlanning webservices URL'),

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

    )

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('eSirius')

    def request(self, uri, method='get', payload=None):

        url = urljoin(self.base_url, uri)

        des_key = pad(force_bytes(self.secret_key), 8)[:8]

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

        epoch = int(datetime.now().strftime('%s%f')[:-3])

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

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

        headers = {

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

            'token_info_caller': base64.b64encode(msg),

        }

        if method == 'get':

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

        elif method == 'post':

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

        else:

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

            if response.status_code == 304:

                raise APIError('Booking not found')

        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},

            )

        return response

    def check_status(self):

        """

        Raise an exception if something goes wrong.

        """

        self.request('sites/')

    @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['user'] = {}

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

            post_data['user']['address'] = {}

        response = self.request('appointments/', 'post', payload=post_data)

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

    @endpoint(

        display_category=_('Appointment'),

        name='get-appointment',

        perm='can_access',

        methods=['get'],

        description=_('Get appointment'),

        parameters={

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

        },

    )

    def get_appointment(self, request, booking_id):

        response = self.request('appointments/%s/' % booking_id)

        return {'data': response.json()}

    @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):

        response = self.request('appointments/%s/' % booking_id, 'delete')

        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 json

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

    },

}

GET_APPOINTMENT_RESPONSE = '''

{

   "beginDate" : "2021-02-26",

   "beginTime" : "16:40",

   "codeRDV" : "943A98",

   "comment" : "coucou",

   "endDate" : "2021-02-26",

   "endTime" : "17:00",

   "idSys" : 108840,

   "isoLanguage" : "fr",

   "motives" : [],

   "needsConfirmation" : false,

   "rdvChannel" : "EAPP0",

   "receptionChannel" : "WS",

   "resources" : {

      "id" : 29,

      "key" : "46",

      "name" : "C1",

      "type" : "STATION"

   },

   "serviceId" : "39",

   "siteCode" : "site1",

   "siteIdSys" : 5,

   "user" : {

      "additionalPersonalIdentity" : [],

      "address" : {},

      "civility" : "",

      "idSys" : "95897"

   }

}

'''

@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)

@pytest.mark.freeze_time('2021-01-26 15:13:6.880')

def test_token(connector):

    connector.secret_id = 'eAppointment'

    connector.secret_key = 'ES2I Info Caller Http Encryption Key'

    connector.save()

    @httmock.all_requests

    def sigerly_mock(url, request):

        assert (

            request.headers['token_info_caller']

            == b'yM4zYAxT67Qvjd20riG3j0eu0t0Ku+HLlttj17Gul7zkruFaXX1J8BJ6sV2Ldgw40axfWh+ESAY='

        )

        return httmock.response(200)

    with httmock.HTTMock(sigerly_mock):

        connector.request('an/uri/', payload="somes")

def test_pre_request(connector):

    @httmock.urlmatch(netloc='dummy-server.org', path='/an/uri/', method='GET')

    def sigerly_mock(url, request):

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

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

        return httmock.response(200)

    with httmock.HTTMock(sigerly_mock):

        connector.request('an/uri/', payload="somes")

@pytest.mark.parametrize(

    'status_code, content, a_dict',

    [

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

        (500, 'not json', None),

    ],

)

def test_post_request(connector, status_code, content, a_dict):

    @httmock.urlmatch(netloc='dummy-server.org', path='/an/uri/', method='GET')

    def sigerly_mock(url, request):

        return httmock.response(status_code, content)

    with pytest.raises(APIError) as exc:

        with httmock.HTTMock(sigerly_mock):

            connector.request('an/uri/', payload="somes")

    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_get_appointment(app, connector):

    endpoint = get_endpoint('get-appointment')

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

    def sigerly_mock(url, request):

        return httmock.response(200, GET_APPOINTMENT_RESPONSE)

    with httmock.HTTMock(sigerly_mock):

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

    assert not resp.json['err']

    assert resp.json['data']['codeRDV']

    assert resp.json['data'] == json.loads(GET_APPOINTMENT_RESPONSE)

def test_get_appointment_error(app, connector):

    endpoint = get_endpoint('get-appointment')

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

    def sigerly_mock(url, request):

        return httmock.response(404, '{"code":"Not Found","message":"Le rendez-vous {0} n\'existe pas"}')

    with httmock.HTTMock(sigerly_mock):

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

    assert resp.json['err']

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

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

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

        "code": "Not Found",

        "message": "Le rendez-vous {0} n'existe pas",

    }

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'