entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 15 Releases Activity

signal_arretes: create signal arretes connector (#65822)

This commit is contained in:
Corentin Sechet 2022-06-01 15:12:51 +02:00
parent 9d9c6f0ab6
commit 2b14a8d6c6
6 changed files with 668 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
passerelle/apps/signal_arretes/__init__.py Normal file
View File

70
passerelle/apps/signal_arretes/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,70 @@
# 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 ™',
},
),
]

0
passerelle/apps/signal_arretes/migrations/__init__.py Normal file
View File

295
passerelle/apps/signal_arretes/models.py Normal file
View File

@ -0,0 +1,295 @@
# -*- 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

1
passerelle/settings.py
View File

@ -164,6 +164,7 @@ INSTALLED_APPS = (
'passerelle.apps.plone_restapi',
'passerelle.apps.sector',
'passerelle.apps.sfr_dmc',
'passerelle.apps.signal_arretes',
'passerelle.apps.sivin',
'passerelle.apps.soap',
'passerelle.apps.solis',

302
tests/test_signal_arretes.py Normal file
View File

@ -0,0 +1,302 @@
# 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']