Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.26 on 2022-06-13 14:13

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='SignalArretes',

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

                ('base_url', models.URLField(verbose_name='Base API URL')),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_signalarretes_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'Signal Arrêtés',

            },

        ),

    ]

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

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

import json

import re

from base64 import b64decode

from datetime import datetime

from django.db import models

from django.http import HttpResponse

from django.utils.text import slugify

from django.utils.translation import ugettext_lazy as _

from requests import RequestException

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

REQUEST_SCHEMA = {

    'type': 'object',

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

    'title': 'Signal Arretes',

    'description': 'Public Occupation Request Schema',

    'required': [

        'declarant_organisation',

        'declarant_quality',

        'declarant_civility',

        'declarant_name',

        'declarant_surname',

        'declarant_email',

        'occupation_lane',

        'occupation_city',

        'occupation_type',

        'occupation_start_date',

        'occupation_end_date',

    ],

    'properties': {

        'declarant_organisation': {

            'description': _('"Individual" or enterprise name'),

            'type': 'string',

        },

        'declarant_siret': {

            'description': _('Entreprise SIRET number'),

            'type': ['string', 'null'],

            'pattern': '\\d{14}',

            'pattern_description': _('14-digits siret number'),

        },

        'declarant_quality': {

            'description': _('Declarant quality'),

            'type': 'string',

            'enum': ['Particulier', 'Entreprise', 'Association'],

        },

        'file_number': {'description': _('Declarant reference'), 'type': ['string', 'null']},

        'declarant_civility': {

            'description': _('Declarant civility'),

            'type': 'string',

            'enum': ['MONSIEUR', 'MADAME'],

        },

        'declarant_name': {'description': _('Declarant name'), 'type': 'string'},

        'declarant_surname': {'description': _('Declarant surname'), 'type': 'string'},

        'declarant_address': {'description': _('Declarant address'), 'type': ['string', 'null']},

        'declarant_zip': {'description': _('Declarant ZIP code'), 'type': ['string', 'null']},

        'declarant_city': {'description': _('Declarant city'), 'type': ['string', 'null']},

        'declarant_email': {'description': _('Declarant email address'), 'type': 'string'},

        'declarant_phone': {'description': _('Declarant phone number'), 'type': ['string', 'null']},

        'occupation_lane': {'description': _('Occupation lane'), 'type': 'string'},

        'occupation_number': {'description': _('Occupation lane number'), 'type': ['string', 'null']},

        'occupation_city': {'description': _('Occupation city'), 'type': 'string'},

        'occupation_type': {'description': _('Occupation type'), 'type': 'string'},

        'occupation_start_date': {

            'description': _('Occupation start date'),

            'type': 'string',

            'format': 'date',

        },

        'occupation_end_date': {'description': _('Occupation end date'), 'type': 'string', 'format': 'date'},

    },

}

class SignalArretes(BaseResource):

    base_url = models.URLField(_('Base API URL'))

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = 'Signal Arrêtés'

    @classmethod

    def get_verbose_name(cls):

        return cls._meta.verbose_name

    @endpoint(

        description=_('Create a public domain occupation request'),

        perm='can_access',

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

    )

    def create_request(self, request, post_data):

        def _format_date(date_string):

            return datetime.strptime(date_string, '%d/%m/%Y').strftime('%Y-%m-%d')

        query_data = {

            'organisationDeclarante': post_data['declarant_organisation'],

            'qualite': post_data['declarant_quality'],

            'contact': {

                'civilite': post_data['declarant_civility'],

                'nom': post_data['declarant_name'],

                'prenom': post_data['declarant_surname'],

                'email': post_data['declarant_email'],

            },

            'localisation': {

                'nomVoie': post_data['occupation_lane'],

                'commune': post_data['occupation_city'],

                'natureOccupation': post_data['occupation_type'],

                'dateDebut': _format_date(post_data['occupation_start_date']),

                'dateFin': _format_date(post_data['occupation_end_date']),

            },

        }

        def _update_data(key, target, target_key):

            if key in post_data and post_data[key]:

                target[target_key] = post_data[key]

        _update_data('declarant_siret', query_data, 'SIRET')

        _update_data('file_number', query_data, 'numeroDossier')

        _update_data('declarant_address', query_data['contact'], 'adresseLigne1')

        _update_data('declarant_zip', query_data['contact'], 'CP')

        _update_data('declarant_city', query_data['contact'], 'ville')

        _update_data('declarant_phone', query_data['contact'], 'telephone')

        _update_data('occupation_number', query_data['localisation'], 'numeroVoie')

        result_string = self._call('CreationDODP', query_data)

        if not isinstance(result_string, str):

            raise APIError('Expected a string')

        try:

            result = json.loads(result_string)

        except ValueError:

            raise APIError('Returned string should be valid json')

        if not isinstance(result, dict) or len(result) != 1:

            raise APIError('Expected a dictionary with one element')

        return {'request_id': list(result.keys())[0]}

    @endpoint(

        description=_('Get cities available in Signal Arrêtés'),

        perm='can_access',

    )

    def get_cities(self, request, **kwargs):

        return self._get_list('GetCommunes')

    @endpoint(

        description=_('Get lanes available in Signal Arrêtés'),

        perm='can_access',

        parameters={

            'city': {'description': _('Get lanes for this city')},

        },

    )

    def get_lanes(self, request, city):

        return self._get_list('GetVoies', {'Commune': city})

    @endpoint(

        description=_('Get available occupation types in Signal Arrêtés'),

        perm='can_access',

    )

    def get_occupation_types(self, request):

        return self._get_list('GetNaturesOccupation')

    @endpoint(

        description=_('Get status of given request in signal arrêtés'),

        perm='can_access',

        parameters={

            'request_id': {'description': _('The occupation request id returned by create_request')},

        },

    )

    def get_request_status(self, request, request_id):

        return {'request_status': self._get_value('GetStatutDemande', request_id=request_id)}

    @endpoint(

        description=_('Get document associated with given request in signal arrêtés'),

        perm='can_access',

        parameters={

            'request_id': {'description': _('The occupation request id returned by create_request')},

        },

    )

    def get_request_document(self, request, request_id):

        result = self._get_value('GetDocumentDemande', request_id=request_id)

        filename = result['name']

        content_type = result['contentType']

        try:

            content = b64decode(result['content'], validate=True)

        except binascii.Error as e:

            raise APIError(f'Corrupted base64 content {e}')

        response = HttpResponse(content, content_type=content_type)

        response['Content-Disposition'] = f'attachment; filename="{filename}"'

        return response

    def _call(self, endpoint, post_data=None):

        url = f'{self.base_url}/CreationDemandeService.svc/{endpoint}'

        if not post_data:

            response = self.requests.get(url)

        else:

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

        try:

            response.raise_for_status()

        except RequestException as e:

            if response.status_code == 400:

                error_msg_match = re.search(

                    'Le message d\'exception est \'(.*)\'\\. Pour plus d\'informations', response.text

                )

                if error_msg_match:

                    error_message = error_msg_match.group(1)

                    raise APIError(

                        'An error occured during the request to Signal Arrêtés: %s' % error_message

                    )

            raise APIError('An error occured during the request to Signal Arrêtés: %s' % e)

        try:

            return response.json()

        except ValueError:

            raise APIError('Expected valid json')

    def _get_value(self, endpoint, post_data=None, request_id=None):

        if request_id:

            url = f"{endpoint}/{request_id}"

        else:

            url = endpoint

        response = self._call(url, post_data=post_data)

        result_key = f'{endpoint}Result'

        if not isinstance(response, dict) or result_key not in response:

            raise APIError('Expected a dictionary with a %s key' % result_key)

        result_str = response[result_key]

        try:

            return json.loads(result_str)

        except ValueError:

            raise APIError('Expected valid json string at %s key' % result_key)

    def _get_list(self, endpoint, post_data=None):

        result = self._get_value(endpoint, post_data=post_data)

        if not isinstance(result, list):

            raise APIError('Expected a list')

        return {'data': [{'id': slugify(it), 'text': it} for it in result]}

    'passerelle.apps.signal_arretes',

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

from base64 import standard_b64encode

import httmock

import pytest

from passerelle.apps.signal_arretes.models import SignalArretes

from tests.utils import generic_endpoint_url, setup_access_rights

@pytest.fixture()

def connector(db):

    return setup_access_rights(

        SignalArretes.objects.create(slug='test', base_url='http://test.signalarretes.net')

    )

REQUIRED_PARAMETERS = {

    'declarant_organisation': 'ACME',

    'declarant_siret': None,

    'declarant_quality': 'Entreprise',

    'file_number': None,

    'declarant_civility': 'MADAME',

    'declarant_name': 'John',

    'declarant_surname': 'Doe',

    'declarant_address': None,

    'declarant_zip': None,

    'declarant_city': None,

    'declarant_email': 'john@doe.net',

    'declarant_phone': None,

    'occupation_lane': 'Sesame Street',

    'occupation_number': '10',

    'occupation_city': 'Melun',

    'occupation_type': 'Base de vie',

    'occupation_start_date': '02/06/2022',

    'occupation_end_date': '03/06/2022',

}

ALL_PARAMETERS = dict(REQUIRED_PARAMETERS)

ALL_PARAMETERS.update(

    {

        'declarant_siret': '00000000000000',

        'file_number': 'reference_dossier',

        'declarant_address': '6 Proudhon street',

        'declarant_zip': '42 42420',

        'declarant_city': 'Melun',

        'declarant_phone': '0636656565',

        'occupation_lane': 'Sesame Street',

    }

)

def _check_required_parameters(data):

    assert data['organisationDeclarante'] == 'ACME'

    assert data['qualite'] == 'Entreprise'

    assert data['contact']['civilite'] == 'MADAME'

    assert data['contact']['nom'] == 'John'

    assert data['contact']['prenom'] == 'Doe'

    assert data['contact']['email'] == 'john@doe.net'

    assert data['localisation']['nomVoie'] == 'Sesame Street'

    assert data['localisation']['commune'] == 'Melun'

    assert data['localisation']['natureOccupation'] == 'Base de vie'

    assert data['localisation']['dateDebut'] == '2022-06-02'

    assert data['localisation']['dateFin'] == '2022-06-03'

    assert data['localisation']['numeroVoie'] == '10'

def test_create_request(app, connector):

    @httmock.all_requests

    def create_request_mock(url, request):

        assert request.path_url == '/CreationDemandeService.svc/CreationDODP'

        data = json.loads(request.body)

        _check_required_parameters(data)

        assert data['SIRET'] == '00000000000000'

        assert data['numeroDossier'] == 'reference_dossier'

        assert data['contact']['adresseLigne1'] == '6 Proudhon street'

        assert data['contact']['CP'] == '42 42420'

        assert data['contact']['ville'] == 'Melun'

        assert data['contact']['telephone'] == '0636656565'

        result_json = json.dumps({'D0000_DOT': 'Enregistré'})

        return httmock.response(200, json.dumps(result_json))

    endpoint = generic_endpoint_url('signal-arretes', 'create_request', slug=connector.slug)

    with httmock.HTTMock(create_request_mock):

        response = app.post_json(endpoint, params=ALL_PARAMETERS)

    result = response.json

    assert result['request_id'] == 'D0000_DOT'

def test_create_request_optional_parameters(app, connector):

    @httmock.all_requests

    def create_request_mock(url, request):

        data = json.loads(request.body)

        _check_required_parameters(data)

        assert 'SIRET' not in data

        assert 'numeroDossier' not in data

        assert 'adresseLigne1' not in data['contact']

        assert 'CP' not in data['contact']

        assert 'ville' not in data['contact']

        assert 'telephone' not in data['contact']

        result_json = json.dumps({'D0000_DOT': 'Enregistré'})

        return httmock.response(200, json.dumps(result_json))

    endpoint = generic_endpoint_url('signal-arretes', 'create_request', slug=connector.slug)

    with httmock.HTTMock(create_request_mock):

        response = app.post_json(

            endpoint,

            params=REQUIRED_PARAMETERS,

        )

    result = response.json

    assert result['request_id'] == 'D0000_DOT'

def test_create_request_errors(app, connector):

    def _test_error(response, error_message):

        @httmock.all_requests

        def bad_json_response(url, request):

            return httmock.response(200, response)

        endpoint = generic_endpoint_url('signal-arretes', 'create_request', slug=connector.slug)

        with httmock.HTTMock(bad_json_response):

            response = app.post_json(endpoint, params=REQUIRED_PARAMETERS)

        json_response = response.json

        assert 'err' in json_response

        assert json_response['err']

        assert error_message in json_response['err_desc']

    _test_error('Not valid json', 'Expected valid json')

    _test_error('[]', 'Expected a string')

    _test_error('"Invalid json"', 'Returned string should be valid json')

    _test_error('"[]"', 'Expected a dictionary with one element')

    _test_error('"{}"', 'Expected a dictionary with one element')

def test_get_cities(app, connector):

    @httmock.all_requests

    def get_cities_mock(url, request):

        assert request.path_url == '/CreationDemandeService.svc/GetCommunes'

        result_json = json.dumps(["Clapotis Les Canards", "Grosboule Les Bains"])

        return httmock.response(200, json.dumps({'GetCommunesResult': result_json}))

    endpoint = generic_endpoint_url('signal-arretes', 'get_cities', slug=connector.slug)

    with httmock.HTTMock(get_cities_mock):

        response = app.get(endpoint)

    assert response.json['data'] == [

        {'id': 'clapotis-les-canards', 'text': 'Clapotis Les Canards'},

        {'id': 'grosboule-les-bains', 'text': 'Grosboule Les Bains'},

    ]

def test_get_lanes(app, connector):

    @httmock.all_requests

    def get_lanes_mock(url, request):

        assert request.path_url == '/CreationDemandeService.svc/GetVoies'

        data = json.loads(request.body)

        assert data == {'Commune': 'Clapotis Les Canards'}

        result_json = json.dumps(["Rue Sacco", "Rue Vanzetti"])

        return httmock.response(200, json.dumps({'GetVoiesResult': result_json}))

    endpoint = generic_endpoint_url('signal-arretes', 'get_lanes', slug=connector.slug)

    with httmock.HTTMock(get_lanes_mock):

        response = app.get(endpoint, params={'city': 'Clapotis Les Canards'})

    assert response.json['data'] == [

        {'id': 'rue-sacco', 'text': 'Rue Sacco'},

        {'id': 'rue-vanzetti', 'text': 'Rue Vanzetti'},

    ]

def test_get_occupation_types(app, connector):

    @httmock.all_requests

    def get_occupation_types_mock(url, request):

        assert request.path_url == '/CreationDemandeService.svc/GetNaturesOccupation'

        result_json = json.dumps(["Déménagement", "Festival"])

        return httmock.response(200, json.dumps({'GetNaturesOccupationResult': result_json}))

    endpoint = generic_endpoint_url('signal-arretes', 'get_occupation_types', slug=connector.slug)

    with httmock.HTTMock(get_occupation_types_mock):

        response = app.get(endpoint)

    assert response.json['data'] == [

        {'id': 'demenagement', 'text': 'Déménagement'},

        {'id': 'festival', 'text': 'Festival'},

    ]

def test_get_request_status(app, connector):

    @httmock.all_requests

    def get_occupation_types_mock(url, request):

        assert request.path_url == '/CreationDemandeService.svc/GetStatutDemande/test_request_id'

        result_json = json.dumps("Enregistré")

        return httmock.response(200, json.dumps({'GetStatutDemandeResult': result_json}))

    endpoint = generic_endpoint_url('signal-arretes', 'get_request_status', slug=connector.slug)

    with httmock.HTTMock(get_occupation_types_mock):

        response = app.get(endpoint, params={'request_id': 'test_request_id'})

    assert not response.json['err']

    assert response.json['request_status'] == 'Enregistré'

def test_get_request_document(app, connector):

    test_content = 'Test file content'.encode('utf-8')

    @httmock.all_requests

    def get_document_mock(url, request):

        assert request.path_url == '/CreationDemandeService.svc/GetDocumentDemande/test_request_id'

        content = standard_b64encode(test_content).decode('utf-8')

        result_json = json.dumps(

            {'contentType': 'text/test-data', 'name': 'test_filename', 'content': content}

        )

        return httmock.response(200, json.dumps({'GetDocumentDemandeResult': result_json}))

    endpoint = generic_endpoint_url('signal-arretes', 'get_request_document', slug=connector.slug)

    with httmock.HTTMock(get_document_mock):

        response = app.get(endpoint, params={'request_id': 'test_request_id'})

    assert response.headers['Content-Type'] == 'text/test-data'

    assert response.headers['Content-Disposition'] == 'attachment; filename="test_filename"'

    assert response.body == test_content

    @httmock.all_requests

    def corrupted_content_mock(url, request):

        result_json = json.dumps(

            {'contentType': 'text/test-data', 'name': 'test_filename', 'content': '$$$$$$$'}

        )

        return httmock.response(200, json.dumps({'GetDocumentDemandeResult': result_json}))

    with httmock.HTTMock(corrupted_content_mock):

        response = app.get(endpoint, params={'request_id': 'test_request_id'})

    json_response = response.json

    assert 'err' in json_response

    assert json_response['err']

    assert 'Corrupted base64 content' in json_response['err_desc']

def test_error_handling(app, connector):

    def _test_error(status_code, body, expected_message):

        @httmock.all_requests

        def error_mock(url, request):

            return httmock.response(status_code, body)

        endpoint = generic_endpoint_url('signal-arretes', 'get_cities', slug=connector.slug)

        with httmock.HTTMock(error_mock):

            response = app.get(endpoint)

        json_response = response.json

        assert 'err' in json_response

        assert json_response['err']

        assert expected_message in json_response['err_desc']

    _test_error(

        400,

        'Le message d\'exception est \'Test error message\'. Pour plus d\'informations thanks a lot to use HTML as return of a json api.',

        'An error occured during the request to Signal Arrêtés: Test error message',

    )

    _test_error(500, 'Unmatched message', 'An error occured during the request to Signal Arrêtés')

    _test_error(200, 'Invalid json', 'Expected valid json')

    _test_error(200, '[]', 'Expected a dictionary with a GetCommunesResult key')

    _test_error(200, '{}', 'Expected a dictionary with a GetCommunesResult key')

    _test_error(

        200, '{"GetCommunesResult": "Invalid json"}', 'Expected valid json string at GetCommunesResult key'

    )

    _test_error(200, '{"GetCommunesResult": "{}"}', 'Expected a list')