Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.26 on 2022-06-17 12:25

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

                (

                    '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(blank=True, default=True, verbose_name='TLS verify certificates'),

                ),

                (

                    'http_proxy',

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

                ),

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

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

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

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

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

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

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

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

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

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

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

    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

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

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

        try:

            if not post_data:

                response = self.requests.get(url)

            else:

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

            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, q=None, id=None):

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

        if not isinstance(result, list):

            raise APIError('Expected a list')

        if q is not None:

            q = q.lower()

            result = filter(lambda it: q in it.lower(), result)

        if id is not None:

            result = list(filter(lambda it: slugify(it) == id, result))

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

    @endpoint(

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

        perm='can_access',

        parameters={

            'id': {

                'description': _('Get exactly one city from it\'s id'),

                'example_value': 'base-de-vie',

            },

            'q': {'description': _('Search text'), 'example_value': 'Angou'},

        },

    )

    def cities(self, request, q=None, id=None, **kwargs):

        return self._get_list('GetCommunes', post_data=None, q=q, id=id)

    @endpoint(

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

        perm='can_access',

        parameters={

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

            'id': {

                'description': _('Get exactly one lane from it\'s id'),

                'example_value': 'rue-nicolas-appert',

            },

            'q': {'description': _('Search text'), 'example_value': 'Rue Nic'},

        },

    )

    def lanes(self, request, city, q=None, id=None):

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

    @endpoint(

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

        perm='can_access',

        parameters={

            'id': {

                'description': _('Get exactly one occupation type from it\'s id'),

                'example_value': 'base-de-vie',

            },

            'q': {'description': _('Search text'), 'example_value': 'Base de'},

        },

    )

    def occupation_types(self, request, q=None, id=None):

        return self._get_list('GetNaturesOccupation', q=q, id=id)

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

            'SIRET': post_data['declarant_siret'],

            'numeroDossier': post_data['file_number'],

            'contact': {

                'civilite': post_data['declarant_civility'],

                'nom': post_data['declarant_name'],

                'prenom': post_data['declarant_surname'],

                'email': post_data['declarant_email'],

                'adresseLigne1': post_data['declarant_address'],

                'CP': post_data['declarant_zip'],

                'ville': post_data['declarant_city'],

                'telephone': post_data['declarant_phone'],

            },

            'localisation': {

                'nomVoie': post_data['occupation_lane'],

                'commune': post_data['occupation_city'],

                'natureOccupation': post_data['occupation_type'],

                'dateDebut': datetime.strptime(post_data['occupation_start_date'], '%d/%m/%Y').strftime(

                    '%Y-%m-%d'

                ),

                'dateFin': datetime.strptime(post_data['occupation_end_date'], '%d/%m/%Y').strftime(

                    '%Y-%m-%d'

                ),

                'numeroVoie': post_data['occupation_number'],

            },

        }

        query_data = {k: v for k, v in query_data.items() if v}

        query_data['contact'] = {k: v for k, v in query_data['contact'].items() if v}

        query_data['localisation'] = {k: v for k, v in query_data['localisation'].items() if v}

        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], 'request_status': list(result.values())[0]}

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

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

from httmock import HTTMock, response, urlmatch

from mock import patch

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://sa.net'))

@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetCommunes')

def mock_get_communes(url, request):

    return response(

        200, json.dumps({'GetCommunesResult': json.dumps(['Clapotis Les Canards', 'Grosboule Les Bains'])})

    )

@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetVoies')

def mock_get_voies(url, request):

    assert json.loads(request.body) == {'Commune': 'Clapotis Les Canards'}

    return response(200, json.dumps({'GetVoiesResult': json.dumps(['Rue Sacco', 'Rue Vanzetti'])}))

@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetNaturesOccupation')

def mock_get_natures_occupation(url, request):

    return response(200, json.dumps({'GetNaturesOccupationResult': json.dumps(['Déménagement', 'Festival'])}))

@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/CreationDODP')

def mock_creation_dodp(url, request):

    data = json.loads(request.body)

    contact = data['contact']

    localization = data['localisation']

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

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

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

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

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

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

    assert localization['nomVoie'] == 'Sesame Street'

    assert localization['commune'] == 'Melun'

    assert localization['natureOccupation'] == 'Base de vie'

    assert localization['dateDebut'] == '2022-06-02'

    assert localization['dateFin'] == '2022-06-03'

    assert 'SIRET' not in data or data['SIRET'] == '00000000000000'

    assert 'numeroDossier' not in data or data['numeroDossier'] == 'reference_dossier'

    assert 'adresseLigne1' not in contact or contact['adresseLigne1'] == '6 Sesame street'

    assert 'CP' not in contact or contact['CP'] == '42 42420'

    assert 'ville' not in contact or contact['ville'] == 'Melun'

    assert 'telephone' not in contact or contact['telephone'] == '0636656565'

    assert 'numeroVoie' not in localization or localization['numeroVoie'] == '10'

    return response(200, json.dumps(json.dumps({'D0000_DOT': 'Enregistré'})))

@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetStatutDemande/test_request_id')

def mock_get_statut_demande(url, request):

    return response(200, json.dumps({'GetStatutDemandeResult': json.dumps('Enregistré')}))

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

@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetDocumentDemande/.*')

def mock_get_document_demande(url, request):

    if url.path.endswith('corrupted'):

        content = '$$$$$$$'

    else:

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

    return response(

        200,

        json.dumps(

            {

                'GetDocumentDemandeResult': json.dumps(

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

                )

            }

        ),

    )

@pytest.fixture

def mock_signal_arretes():

    with HTTMock(

        mock_creation_dodp,

        mock_get_communes,

        mock_get_voies,

        mock_get_natures_occupation,

        mock_get_statut_demande,

        mock_get_document_demande,

    ) as mock:

        yield mock

def test_cities(app, connector, mock_signal_arretes):

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

    result = app.get(endpoint)

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

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

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

    ]

    result = app.get(endpoint, params={'id': 'grosboule-les-bains'})

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

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

    ]

    result = app.get(endpoint, params={'q': 'CANARD'})

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

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

    ]

    result = app.get(endpoint, params={'q': 'CANARD', 'id': 'grosboule-les-bains'})

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

def test_lanes(app, connector, mock_signal_arretes):

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

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

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

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

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

    ]

    result = app.get(endpoint, params={'city': 'Clapotis Les Canards', 'id': 'rue-sacco'})

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

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

    ]

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

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

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

    ]

    result = app.get(endpoint, params={'city': 'Clapotis Les Canards', 'q': 'VAN', 'id': 'rue-sacco'})

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

def test_occupation_types(app, connector, mock_signal_arretes):

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

    result = app.get(endpoint)

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

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

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

    ]

    result = app.get(endpoint, params={'id': 'demenagement'})

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

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

    ]

    result = app.get(endpoint, params={'q': 'esti'})

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

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

    ]

    result = app.get(endpoint, params={'q': 'esti', 'id': 'demenagement'})

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

REQUIRED_PARAMETERS = {

    'declarant_organisation': 'ACME',

    'declarant_siret': '',

    'declarant_quality': 'Entreprise',

    'file_number': '',

    'declarant_civility': 'MADAME',

    'declarant_name': 'John',

    'declarant_surname': 'Doe',

    'declarant_address': '',

    'declarant_zip': '',

    'declarant_city': '',

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

    'declarant_phone': '',

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

}

def test_create_request(app, connector, mock_signal_arretes):

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

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

    assert result.json == {'request_id': 'D0000_DOT', 'request_status': 'Enregistré', 'err': 0}

    all_parameters = dict(REQUIRED_PARAMETERS)

    all_parameters.update(

        {

            'declarant_siret': '00000000000000',

            'file_number': 'reference_dossier',

            'declarant_address': '6 Sesame street',

            'declarant_zip': '42 42420',

            'declarant_city': 'Melun',

            'declarant_phone': '0636656565',

            'occupation_lane': 'Sesame Street',

        }

    )

    result = app.post_json(endpoint, params=all_parameters)

    assert result.json == {'request_id': 'D0000_DOT', 'request_status': 'Enregistré', 'err': 0}

@patch('passerelle.utils.Request.post')

@pytest.mark.parametrize(

    'response_body,error_message',

    [

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

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

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

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

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

    ],

)

def test_create_request_errors(mocked_post, app, connector, response_body, error_message):

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

    mocked_post.return_value = response(200, response_body)

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

    assert 'err' in result.json

    assert result.json['err']

    assert error_message in result.json['err_desc']

def test_request_status(app, connector, mock_signal_arretes):

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

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

    assert not result.json['err']

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

def test_request_document(app, connector, mock_signal_arretes):

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

    result = app.get(endpoint, params={'request_id': 'document'})

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

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

    assert result.body == DOCUMENT_CONTENT

    result = app.get(endpoint, params={'request_id': 'document_corrupted'})

    assert 'err' in result.json

    assert result.json['err']

    assert 'Corrupted base64 content' in result.json['err_desc']

@patch('passerelle.utils.Request.get')

@pytest.mark.parametrize(

    'status_code,body,expected_message',

    [

        (

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

        ),

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

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

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

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

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

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

    ],

)

def test_error_handling(mocked_get, app, connector, status_code, body, expected_message):

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

    mocked_get.return_value = response(status_code, body)

    result = app.get(endpoint)

    assert 'err' in result.json

    assert result.json['err']

    assert expected_message in result.json['err_desc']