0001-signal_arretes-create-signal-arretes-connector-65822.patch
passerelle/apps/signal_arretes/migrations/0001_initial.py | ||
---|---|---|
1 |
# Generated by Django 2.2.26 on 2022-06-17 12:25 |
|
2 | ||
3 |
from django.db import migrations, models |
|
4 | ||
5 | ||
6 |
class Migration(migrations.Migration): |
|
7 | ||
8 |
initial = True |
|
9 | ||
10 |
dependencies = [ |
|
11 |
('base', '0029_auto_20210202_1627'), |
|
12 |
] |
|
13 | ||
14 |
operations = [ |
|
15 |
migrations.CreateModel( |
|
16 |
name='SignalArretes', |
|
17 |
fields=[ |
|
18 |
( |
|
19 |
'id', |
|
20 |
models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'), |
|
21 |
), |
|
22 |
('title', models.CharField(max_length=50, verbose_name='Title')), |
|
23 |
('slug', models.SlugField(unique=True, verbose_name='Identifier')), |
|
24 |
('description', models.TextField(verbose_name='Description')), |
|
25 |
( |
|
26 |
'basic_auth_username', |
|
27 |
models.CharField( |
|
28 |
blank=True, max_length=128, verbose_name='Basic authentication username' |
|
29 |
), |
|
30 |
), |
|
31 |
( |
|
32 |
'basic_auth_password', |
|
33 |
models.CharField( |
|
34 |
blank=True, max_length=128, verbose_name='Basic authentication password' |
|
35 |
), |
|
36 |
), |
|
37 |
( |
|
38 |
'client_certificate', |
|
39 |
models.FileField( |
|
40 |
blank=True, null=True, upload_to='', verbose_name='TLS client certificate' |
|
41 |
), |
|
42 |
), |
|
43 |
( |
|
44 |
'trusted_certificate_authorities', |
|
45 |
models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'), |
|
46 |
), |
|
47 |
( |
|
48 |
'verify_cert', |
|
49 |
models.BooleanField(blank=True, default=True, verbose_name='TLS verify certificates'), |
|
50 |
), |
|
51 |
( |
|
52 |
'http_proxy', |
|
53 |
models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'), |
|
54 |
), |
|
55 |
('base_url', models.URLField(verbose_name='Base API URL')), |
|
56 |
( |
|
57 |
'users', |
|
58 |
models.ManyToManyField( |
|
59 |
blank=True, |
|
60 |
related_name='_signalarretes_users_+', |
|
61 |
related_query_name='+', |
|
62 |
to='base.ApiUser', |
|
63 |
), |
|
64 |
), |
|
65 |
], |
|
66 |
options={ |
|
67 |
'verbose_name': 'Signal Arrêtés ™', |
|
68 |
}, |
|
69 |
), |
|
70 |
] |
passerelle/apps/signal_arretes/models.py | ||
---|---|---|
1 |
# -*- coding: utf-8 -*- |
|
2 |
# Copyright (C) 2020 Entr'ouvert |
|
3 |
# |
|
4 |
# This program is free software: you can redistribute it and/or modify it |
|
5 |
# under the terms of the GNU Affero General Public License as published |
|
6 |
# by the Free Software Foundation, either version 3 of the License, or |
|
7 |
# (at your option) any later version. |
|
8 |
# |
|
9 |
# This program is distributed in the hope that it will be useful, |
|
10 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 |
# GNU Affero General Public License for more details. |
|
13 |
# |
|
14 |
# You should have received a copy of the GNU Affero General Public License |
|
15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | ||
17 | ||
18 |
import binascii |
|
19 |
import json |
|
20 |
import re |
|
21 |
from base64 import b64decode |
|
22 |
from datetime import datetime |
|
23 | ||
24 |
from django.db import models |
|
25 |
from django.http import HttpResponse |
|
26 |
from django.utils.text import slugify |
|
27 |
from django.utils.translation import ugettext_lazy as _ |
|
28 |
from requests import RequestException |
|
29 | ||
30 |
from passerelle.base.models import BaseResource, HTTPResource |
|
31 |
from passerelle.utils.api import endpoint |
|
32 |
from passerelle.utils.jsonresponse import APIError |
|
33 | ||
34 |
REQUEST_SCHEMA = { |
|
35 |
'type': 'object', |
|
36 |
'$schema': 'http://json-schema.org/draft-04/schema#', |
|
37 |
'title': 'Signal Arretes', |
|
38 |
'description': 'Public Occupation Request Schema', |
|
39 |
'required': [ |
|
40 |
'declarant_organisation', |
|
41 |
'declarant_quality', |
|
42 |
'declarant_civility', |
|
43 |
'declarant_name', |
|
44 |
'declarant_surname', |
|
45 |
'declarant_email', |
|
46 |
'occupation_lane', |
|
47 |
'occupation_city', |
|
48 |
'occupation_type', |
|
49 |
'occupation_start_date', |
|
50 |
'occupation_end_date', |
|
51 |
], |
|
52 |
'properties': { |
|
53 |
'declarant_organisation': { |
|
54 |
'description': _('"Individual" or enterprise name'), |
|
55 |
'type': 'string', |
|
56 |
}, |
|
57 |
'declarant_siret': { |
|
58 |
'description': _('Entreprise SIRET number'), |
|
59 |
'type': ['string', 'null'], |
|
60 |
'pattern': '\\d{14}', |
|
61 |
'pattern_description': _('14-digits siret number'), |
|
62 |
}, |
|
63 |
'declarant_quality': { |
|
64 |
'description': _('Declarant quality'), |
|
65 |
'type': 'string', |
|
66 |
'enum': ['Particulier', 'Entreprise', 'Association'], |
|
67 |
}, |
|
68 |
'file_number': {'description': _('Declarant reference'), 'type': ['string', 'null']}, |
|
69 |
'declarant_civility': { |
|
70 |
'description': _('Declarant civility'), |
|
71 |
'type': 'string', |
|
72 |
'enum': ['MONSIEUR', 'MADAME'], |
|
73 |
}, |
|
74 |
'declarant_name': {'description': _('Declarant name'), 'type': 'string'}, |
|
75 |
'declarant_surname': {'description': _('Declarant surname'), 'type': 'string'}, |
|
76 |
'declarant_address': {'description': _('Declarant address'), 'type': ['string', 'null']}, |
|
77 |
'declarant_zip': {'description': _('Declarant ZIP code'), 'type': ['string', 'null']}, |
|
78 |
'declarant_city': {'description': _('Declarant city'), 'type': ['string', 'null']}, |
|
79 |
'declarant_email': {'description': _('Declarant email address'), 'type': 'string'}, |
|
80 |
'declarant_phone': {'description': _('Declarant phone number'), 'type': ['string', 'null']}, |
|
81 |
'occupation_lane': {'description': _('Occupation lane'), 'type': 'string'}, |
|
82 |
'occupation_number': {'description': _('Occupation lane number'), 'type': ['string', 'null']}, |
|
83 |
'occupation_city': {'description': _('Occupation city'), 'type': 'string'}, |
|
84 |
'occupation_type': {'description': _('Occupation type'), 'type': 'string'}, |
|
85 |
'occupation_start_date': { |
|
86 |
'description': _('Occupation start date'), |
|
87 |
'type': 'string', |
|
88 |
'format': 'date', |
|
89 |
}, |
|
90 |
'occupation_end_date': {'description': _('Occupation end date'), 'type': 'string', 'format': 'date'}, |
|
91 |
}, |
|
92 |
} |
|
93 | ||
94 | ||
95 |
class SignalArretes(BaseResource, HTTPResource): |
|
96 |
base_url = models.URLField(_('Base API URL')) |
|
97 |
category = _('Business Process Connectors') |
|
98 | ||
99 |
class Meta: |
|
100 |
verbose_name = 'Signal Arrêtés ™' |
|
101 | ||
102 |
@classmethod |
|
103 |
def get_verbose_name(cls): |
|
104 |
return cls._meta.verbose_name |
|
105 | ||
106 |
def _call(self, endpoint, post_data=None): |
|
107 |
url = f'{self.base_url}/CreationDemandeService.svc/{endpoint}' |
|
108 | ||
109 |
try: |
|
110 |
if not post_data: |
|
111 |
response = self.requests.get(url) |
|
112 |
else: |
|
113 |
response = self.requests.post(url, json=post_data) |
|
114 | ||
115 |
response.raise_for_status() |
|
116 |
except RequestException as e: |
|
117 |
if response.status_code == 400: |
|
118 |
error_msg_match = re.search( |
|
119 |
'Le message d\'exception est \'(.*)\'\\. Pour plus d\'informations', response.text |
|
120 |
) |
|
121 |
if error_msg_match: |
|
122 |
error_message = error_msg_match.group(1) |
|
123 |
raise APIError( |
|
124 |
'An error occured during the request to Signal Arrêtés: %s' % error_message |
|
125 |
) |
|
126 | ||
127 |
raise APIError('An error occured during the request to Signal Arrêtés: %s' % e) |
|
128 | ||
129 |
try: |
|
130 |
return response.json() |
|
131 |
except ValueError: |
|
132 |
raise APIError('Expected valid json') |
|
133 | ||
134 |
def _get_value(self, endpoint, post_data=None, request_id=None): |
|
135 |
if request_id: |
|
136 |
url = f"{endpoint}/{request_id}" |
|
137 |
else: |
|
138 |
url = endpoint |
|
139 | ||
140 |
response = self._call(url, post_data=post_data) |
|
141 |
result_key = f'{endpoint}Result' |
|
142 |
if not isinstance(response, dict) or result_key not in response: |
|
143 |
raise APIError('Expected a dictionary with a %s key' % result_key) |
|
144 | ||
145 |
result_str = response[result_key] |
|
146 |
try: |
|
147 |
return json.loads(result_str) |
|
148 |
except ValueError: |
|
149 |
raise APIError('Expected valid json string at %s key' % result_key) |
|
150 | ||
151 |
def _get_list(self, endpoint, post_data=None, q=None, id=None): |
|
152 |
result = self._get_value(endpoint, post_data=post_data) |
|
153 |
if not isinstance(result, list): |
|
154 |
raise APIError('Expected a list') |
|
155 | ||
156 |
if q is not None: |
|
157 |
q = q.lower() |
|
158 |
result = filter(lambda it: q in it.lower(), result) |
|
159 | ||
160 |
if id is not None: |
|
161 |
result = list(filter(lambda it: slugify(it) == id, result)) |
|
162 | ||
163 |
return {'data': [{'id': slugify(it), 'text': it} for it in result]} |
|
164 | ||
165 |
@endpoint( |
|
166 |
description=_('Get cities available in Signal Arrêtés'), |
|
167 |
perm='can_access', |
|
168 |
parameters={ |
|
169 |
'id': { |
|
170 |
'description': _('Get exactly one city from it\'s id'), |
|
171 |
'example_value': 'base-de-vie', |
|
172 |
}, |
|
173 |
'q': {'description': _('Search text'), 'example_value': 'Angou'}, |
|
174 |
}, |
|
175 |
) |
|
176 |
def cities(self, request, q=None, id=None, **kwargs): |
|
177 |
return self._get_list('GetCommunes', post_data=None, q=q, id=id) |
|
178 | ||
179 |
@endpoint( |
|
180 |
description=_('Get lanes available in Signal Arrêtés'), |
|
181 |
perm='can_access', |
|
182 |
parameters={ |
|
183 |
'city': {'description': _('Get lanes for this city')}, |
|
184 |
'id': { |
|
185 |
'description': _('Get exactly one lane from it\'s id'), |
|
186 |
'example_value': 'rue-nicolas-appert', |
|
187 |
}, |
|
188 |
'q': {'description': _('Search text'), 'example_value': 'Rue Nic'}, |
|
189 |
}, |
|
190 |
) |
|
191 |
def lanes(self, request, city, q=None, id=None): |
|
192 |
return self._get_list('GetVoies', {'Commune': city}, q=q, id=id) |
|
193 | ||
194 |
@endpoint( |
|
195 |
description=_('Get available occupation types in Signal Arrêtés'), |
|
196 |
perm='can_access', |
|
197 |
parameters={ |
|
198 |
'id': { |
|
199 |
'description': _('Get exactly one occupation type from it\'s id'), |
|
200 |
'example_value': 'base-de-vie', |
|
201 |
}, |
|
202 |
'q': {'description': _('Search text'), 'example_value': 'Base de'}, |
|
203 |
}, |
|
204 |
) |
|
205 |
def occupation_types(self, request, q=None, id=None): |
|
206 |
return self._get_list('GetNaturesOccupation', q=q, id=id) |
|
207 | ||
208 |
@endpoint( |
|
209 |
description=_('Create a public domain occupation request'), |
|
210 |
perm='can_access', |
|
211 |
post={'request_body': {'schema': {'application/json': REQUEST_SCHEMA}}}, |
|
212 |
) |
|
213 |
def create_request(self, request, post_data): |
|
214 |
def _format_date(date_string): |
|
215 |
return datetime.strptime(date_string, '%d/%m/%Y').strftime('%Y-%m-%d') |
|
216 | ||
217 |
query_data = { |
|
218 |
'organisationDeclarante': post_data['declarant_organisation'], |
|
219 |
'qualite': post_data['declarant_quality'], |
|
220 |
'SIRET': post_data['declarant_siret'], |
|
221 |
'numeroDossier': post_data['file_number'], |
|
222 |
'contact': { |
|
223 |
'civilite': post_data['declarant_civility'], |
|
224 |
'nom': post_data['declarant_name'], |
|
225 |
'prenom': post_data['declarant_surname'], |
|
226 |
'email': post_data['declarant_email'], |
|
227 |
'adresseLigne1': post_data['declarant_address'], |
|
228 |
'CP': post_data['declarant_zip'], |
|
229 |
'ville': post_data['declarant_city'], |
|
230 |
'telephone': post_data['declarant_phone'], |
|
231 |
}, |
|
232 |
'localisation': { |
|
233 |
'nomVoie': post_data['occupation_lane'], |
|
234 |
'commune': post_data['occupation_city'], |
|
235 |
'natureOccupation': post_data['occupation_type'], |
|
236 |
'dateDebut': datetime.strptime(post_data['occupation_start_date'], '%d/%m/%Y').strftime( |
|
237 |
'%Y-%m-%d' |
|
238 |
), |
|
239 |
'dateFin': datetime.strptime(post_data['occupation_end_date'], '%d/%m/%Y').strftime( |
|
240 |
'%Y-%m-%d' |
|
241 |
), |
|
242 |
'numeroVoie': post_data['occupation_number'], |
|
243 |
}, |
|
244 |
} |
|
245 | ||
246 |
query_data = {k: v for k, v in query_data.items() if v} |
|
247 |
query_data['contact'] = {k: v for k, v in query_data['contact'].items() if v} |
|
248 |
query_data['localisation'] = {k: v for k, v in query_data['localisation'].items() if v} |
|
249 | ||
250 |
result_string = self._call('CreationDODP', query_data) |
|
251 |
if not isinstance(result_string, str): |
|
252 |
raise APIError('Expected a string') |
|
253 | ||
254 |
try: |
|
255 |
result = json.loads(result_string) |
|
256 |
except ValueError: |
|
257 |
raise APIError('Returned string should be valid json') |
|
258 | ||
259 |
if not isinstance(result, dict) or len(result) != 1: |
|
260 |
raise APIError('Expected a dictionary with one element') |
|
261 | ||
262 |
return {'request_id': list(result.keys())[0], 'request_status': list(result.values())[0]} |
|
263 | ||
264 |
@endpoint( |
|
265 |
description=_('Get status of given request in Signal Arrêtés'), |
|
266 |
perm='can_access', |
|
267 |
parameters={ |
|
268 |
'request_id': {'description': _('The occupation request id returned by create_request')}, |
|
269 |
}, |
|
270 |
) |
|
271 |
def request_status(self, request, request_id): |
|
272 |
return {'request_status': self._get_value('GetStatutDemande', request_id=request_id)} |
|
273 | ||
274 |
@endpoint( |
|
275 |
description=_('Get document associated with given request in Signal Arrêtés'), |
|
276 |
perm='can_access', |
|
277 |
parameters={ |
|
278 |
'request_id': {'description': _('The occupation request id returned by create_request')}, |
|
279 |
}, |
|
280 |
) |
|
281 |
def request_document(self, request, request_id): |
|
282 |
result = self._get_value('GetDocumentDemande', request_id=request_id) |
|
283 | ||
284 |
filename = result['name'] |
|
285 |
content_type = result['contentType'] |
|
286 | ||
287 |
try: |
|
288 |
content = b64decode(result['content'], validate=True) |
|
289 |
except binascii.Error as e: |
|
290 |
raise APIError(f'Corrupted base64 content {e}') |
|
291 | ||
292 |
response = HttpResponse(content, content_type=content_type) |
|
293 |
response['Content-Disposition'] = f'attachment; filename="{filename}"' |
|
294 | ||
295 |
return response |
passerelle/settings.py | ||
---|---|---|
164 | 164 |
'passerelle.apps.plone_restapi', |
165 | 165 |
'passerelle.apps.sector', |
166 | 166 |
'passerelle.apps.sfr_dmc', |
167 |
'passerelle.apps.signal_arretes', |
|
167 | 168 |
'passerelle.apps.soap', |
168 | 169 |
'passerelle.apps.solis', |
169 | 170 |
'passerelle.apps.sp_fr', |
tests/test_signal_arretes.py | ||
---|---|---|
1 |
# passerelle - uniform access to multiple data sources and services |
|
2 |
# Copyright (C) 2022 Entr'ouvert |
|
3 |
# |
|
4 |
# This program is free software: you can redistribute it and/or modify it |
|
5 |
# under the terms of the GNU Affero General Public License as published |
|
6 |
# by the Free Software Foundation, either version 3 of the License, or |
|
7 |
# (at your option) any later version. |
|
8 |
# |
|
9 |
# This program is distributed in the hope that it will be useful, |
|
10 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 |
# GNU Affero General Public License for more details. |
|
13 |
# |
|
14 |
# You should have received a copy of the GNU Affero General Public License |
|
15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | ||
17 |
import json |
|
18 |
from base64 import standard_b64encode |
|
19 | ||
20 |
import pytest |
|
21 |
from httmock import HTTMock, response, urlmatch |
|
22 |
from mock import patch |
|
23 | ||
24 |
from passerelle.apps.signal_arretes.models import SignalArretes |
|
25 |
from tests.utils import generic_endpoint_url, setup_access_rights |
|
26 | ||
27 | ||
28 |
@pytest.fixture() |
|
29 |
def connector(db): |
|
30 |
return setup_access_rights(SignalArretes.objects.create(slug='test', base_url='http://sa.net')) |
|
31 | ||
32 | ||
33 |
@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetCommunes') |
|
34 |
def mock_get_communes(url, request): |
|
35 |
return response( |
|
36 |
200, json.dumps({'GetCommunesResult': json.dumps(['Clapotis Les Canards', 'Grosboule Les Bains'])}) |
|
37 |
) |
|
38 | ||
39 | ||
40 |
@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetVoies') |
|
41 |
def mock_get_voies(url, request): |
|
42 |
assert json.loads(request.body) == {'Commune': 'Clapotis Les Canards'} |
|
43 |
return response(200, json.dumps({'GetVoiesResult': json.dumps(['Rue Sacco', 'Rue Vanzetti'])})) |
|
44 | ||
45 | ||
46 |
@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetNaturesOccupation') |
|
47 |
def mock_get_natures_occupation(url, request): |
|
48 |
return response(200, json.dumps({'GetNaturesOccupationResult': json.dumps(['Déménagement', 'Festival'])})) |
|
49 | ||
50 | ||
51 |
@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/CreationDODP') |
|
52 |
def mock_creation_dodp(url, request): |
|
53 |
data = json.loads(request.body) |
|
54 |
contact = data['contact'] |
|
55 |
localization = data['localisation'] |
|
56 | ||
57 |
assert data['organisationDeclarante'] == 'ACME' |
|
58 |
assert data['qualite'] == 'Entreprise' |
|
59 |
assert contact['civilite'] == 'MADAME' |
|
60 |
assert contact['nom'] == 'John' |
|
61 |
assert contact['prenom'] == 'Doe' |
|
62 |
assert contact['email'] == 'john@doe.net' |
|
63 |
assert localization['nomVoie'] == 'Sesame Street' |
|
64 |
assert localization['commune'] == 'Melun' |
|
65 |
assert localization['natureOccupation'] == 'Base de vie' |
|
66 |
assert localization['dateDebut'] == '2022-06-02' |
|
67 |
assert localization['dateFin'] == '2022-06-03' |
|
68 | ||
69 |
assert 'SIRET' not in data or data['SIRET'] == '00000000000000' |
|
70 |
assert 'numeroDossier' not in data or data['numeroDossier'] == 'reference_dossier' |
|
71 |
assert 'adresseLigne1' not in contact or contact['adresseLigne1'] == '6 Sesame street' |
|
72 |
assert 'CP' not in contact or contact['CP'] == '42 42420' |
|
73 |
assert 'ville' not in contact or contact['ville'] == 'Melun' |
|
74 |
assert 'telephone' not in contact or contact['telephone'] == '0636656565' |
|
75 |
assert 'numeroVoie' not in localization or localization['numeroVoie'] == '10' |
|
76 | ||
77 |
return response(200, json.dumps(json.dumps({'D0000_DOT': 'Enregistré'}))) |
|
78 | ||
79 | ||
80 |
@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetStatutDemande/test_request_id') |
|
81 |
def mock_get_statut_demande(url, request): |
|
82 |
return response(200, json.dumps({'GetStatutDemandeResult': json.dumps('Enregistré')})) |
|
83 | ||
84 | ||
85 |
DOCUMENT_CONTENT = 'Test file content'.encode('utf-8') |
|
86 | ||
87 | ||
88 |
@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetDocumentDemande/.*') |
|
89 |
def mock_get_document_demande(url, request): |
|
90 |
if url.path.endswith('corrupted'): |
|
91 |
content = '$$$$$$$' |
|
92 |
else: |
|
93 |
content = standard_b64encode(DOCUMENT_CONTENT).decode('utf-8') |
|
94 | ||
95 |
return response( |
|
96 |
200, |
|
97 |
json.dumps( |
|
98 |
{ |
|
99 |
'GetDocumentDemandeResult': json.dumps( |
|
100 |
{'contentType': 'text/test-data', 'name': 'test_filename', 'content': content} |
|
101 |
) |
|
102 |
} |
|
103 |
), |
|
104 |
) |
|
105 | ||
106 | ||
107 |
@pytest.fixture |
|
108 |
def mock_signal_arretes(): |
|
109 |
with HTTMock( |
|
110 |
mock_creation_dodp, |
|
111 |
mock_get_communes, |
|
112 |
mock_get_voies, |
|
113 |
mock_get_natures_occupation, |
|
114 |
mock_get_statut_demande, |
|
115 |
mock_get_document_demande, |
|
116 |
) as mock: |
|
117 |
yield mock |
|
118 | ||
119 | ||
120 |
def test_cities(app, connector, mock_signal_arretes): |
|
121 |
endpoint = generic_endpoint_url('signal-arretes', 'cities', slug=connector.slug) |
|
122 | ||
123 |
result = app.get(endpoint) |
|
124 |
assert result.json['data'] == [ |
|
125 |
{'id': 'clapotis-les-canards', 'text': 'Clapotis Les Canards'}, |
|
126 |
{'id': 'grosboule-les-bains', 'text': 'Grosboule Les Bains'}, |
|
127 |
] |
|
128 | ||
129 |
result = app.get(endpoint, params={'id': 'grosboule-les-bains'}) |
|
130 |
assert result.json['data'] == [ |
|
131 |
{'id': 'grosboule-les-bains', 'text': 'Grosboule Les Bains'}, |
|
132 |
] |
|
133 | ||
134 |
result = app.get(endpoint, params={'q': 'CANARD'}) |
|
135 |
assert result.json['data'] == [ |
|
136 |
{'id': 'clapotis-les-canards', 'text': 'Clapotis Les Canards'}, |
|
137 |
] |
|
138 | ||
139 |
result = app.get(endpoint, params={'q': 'CANARD', 'id': 'grosboule-les-bains'}) |
|
140 |
assert result.json['data'] == [] |
|
141 | ||
142 | ||
143 |
def test_lanes(app, connector, mock_signal_arretes): |
|
144 |
endpoint = generic_endpoint_url('signal-arretes', 'lanes', slug=connector.slug) |
|
145 | ||
146 |
result = app.get(endpoint, params={'city': 'Clapotis Les Canards'}) |
|
147 |
assert result.json['data'] == [ |
|
148 |
{'id': 'rue-sacco', 'text': 'Rue Sacco'}, |
|
149 |
{'id': 'rue-vanzetti', 'text': 'Rue Vanzetti'}, |
|
150 |
] |
|
151 | ||
152 |
result = app.get(endpoint, params={'city': 'Clapotis Les Canards', 'id': 'rue-sacco'}) |
|
153 |
assert result.json['data'] == [ |
|
154 |
{'id': 'rue-sacco', 'text': 'Rue Sacco'}, |
|
155 |
] |
|
156 | ||
157 |
result = app.get(endpoint, params={'city': 'Clapotis Les Canards', 'q': 'VAN'}) |
|
158 |
assert result.json['data'] == [ |
|
159 |
{'id': 'rue-vanzetti', 'text': 'Rue Vanzetti'}, |
|
160 |
] |
|
161 | ||
162 |
result = app.get(endpoint, params={'city': 'Clapotis Les Canards', 'q': 'VAN', 'id': 'rue-sacco'}) |
|
163 |
assert result.json['data'] == [] |
|
164 | ||
165 | ||
166 |
def test_occupation_types(app, connector, mock_signal_arretes): |
|
167 |
endpoint = generic_endpoint_url('signal-arretes', 'occupation_types', slug=connector.slug) |
|
168 | ||
169 |
result = app.get(endpoint) |
|
170 |
assert result.json['data'] == [ |
|
171 |
{'id': 'demenagement', 'text': 'Déménagement'}, |
|
172 |
{'id': 'festival', 'text': 'Festival'}, |
|
173 |
] |
|
174 | ||
175 |
result = app.get(endpoint, params={'id': 'demenagement'}) |
|
176 |
assert result.json['data'] == [ |
|
177 |
{'id': 'demenagement', 'text': 'Déménagement'}, |
|
178 |
] |
|
179 | ||
180 |
result = app.get(endpoint, params={'q': 'esti'}) |
|
181 |
assert result.json['data'] == [ |
|
182 |
{'id': 'festival', 'text': 'Festival'}, |
|
183 |
] |
|
184 | ||
185 |
result = app.get(endpoint, params={'q': 'esti', 'id': 'demenagement'}) |
|
186 |
assert result.json['data'] == [] |
|
187 | ||
188 | ||
189 |
REQUIRED_PARAMETERS = { |
|
190 |
'declarant_organisation': 'ACME', |
|
191 |
'declarant_siret': None, |
|
192 |
'declarant_quality': 'Entreprise', |
|
193 |
'file_number': None, |
|
194 |
'declarant_civility': 'MADAME', |
|
195 |
'declarant_name': 'John', |
|
196 |
'declarant_surname': 'Doe', |
|
197 |
'declarant_address': None, |
|
198 |
'declarant_zip': None, |
|
199 |
'declarant_city': None, |
|
200 |
'declarant_email': 'john@doe.net', |
|
201 |
'declarant_phone': None, |
|
202 |
'occupation_lane': 'Sesame Street', |
|
203 |
'occupation_number': '10', |
|
204 |
'occupation_city': 'Melun', |
|
205 |
'occupation_type': 'Base de vie', |
|
206 |
'occupation_start_date': '02/06/2022', |
|
207 |
'occupation_end_date': '03/06/2022', |
|
208 |
} |
|
209 | ||
210 | ||
211 |
def test_create_request(app, connector, mock_signal_arretes): |
|
212 |
endpoint = generic_endpoint_url('signal-arretes', 'create_request', slug=connector.slug) |
|
213 |
result = app.post_json(endpoint, params=REQUIRED_PARAMETERS) |
|
214 |
assert result.json == {'request_id': 'D0000_DOT', 'request_status': 'Enregistré', 'err': 0} |
|
215 | ||
216 |
all_parameters = dict(REQUIRED_PARAMETERS) |
|
217 |
all_parameters.update( |
|
218 |
{ |
|
219 |
'declarant_siret': '00000000000000', |
|
220 |
'file_number': 'reference_dossier', |
|
221 |
'declarant_address': '6 Sesame street', |
|
222 |
'declarant_zip': '42 42420', |
|
223 |
'declarant_city': 'Melun', |
|
224 |
'declarant_phone': '0636656565', |
|
225 |
'occupation_lane': 'Sesame Street', |
|
226 |
} |
|
227 |
) |
|
228 | ||
229 |
result = app.post_json(endpoint, params=all_parameters) |
|
230 |
assert result.json == {'request_id': 'D0000_DOT', 'request_status': 'Enregistré', 'err': 0} |
|
231 | ||
232 | ||
233 |
@patch('passerelle.utils.Request.post') |
|
234 |
@pytest.mark.parametrize( |
|
235 |
'response_body,error_message', |
|
236 |
[ |
|
237 |
('Not valid json', 'Expected valid json'), |
|
238 |
('[]', 'Expected a string'), |
|
239 |
('"Invalid json"', 'Returned string should be valid json'), |
|
240 |
('"[]"', 'Expected a dictionary with one element'), |
|
241 |
('"{}"', 'Expected a dictionary with one element'), |
|
242 |
], |
|
243 |
) |
|
244 |
def test_create_request_errors(mocked_post, app, connector, response_body, error_message): |
|
245 |
endpoint = generic_endpoint_url('signal-arretes', 'create_request', slug=connector.slug) |
|
246 |
mocked_post.return_value = response(200, response_body) |
|
247 |
result = app.post_json(endpoint, params=REQUIRED_PARAMETERS) |
|
248 | ||
249 |
assert 'err' in result.json |
|
250 |
assert result.json['err'] |
|
251 |
assert error_message in result.json['err_desc'] |
|
252 | ||
253 | ||
254 |
def test_request_status(app, connector, mock_signal_arretes): |
|
255 |
endpoint = generic_endpoint_url('signal-arretes', 'request_status', slug=connector.slug) |
|
256 |
result = app.get(endpoint, params={'request_id': 'test_request_id'}) |
|
257 | ||
258 |
assert not result.json['err'] |
|
259 |
assert result.json['request_status'] == 'Enregistré' |
|
260 | ||
261 | ||
262 |
def test_request_document(app, connector, mock_signal_arretes): |
|
263 |
endpoint = generic_endpoint_url('signal-arretes', 'request_document', slug=connector.slug) |
|
264 | ||
265 |
result = app.get(endpoint, params={'request_id': 'document'}) |
|
266 | ||
267 |
assert result.headers['Content-Type'] == 'text/test-data' |
|
268 |
assert result.headers['Content-Disposition'] == 'attachment; filename="test_filename"' |
|
269 |
assert result.body == DOCUMENT_CONTENT |
|
270 | ||
271 |
result = app.get(endpoint, params={'request_id': 'document_corrupted'}) |
|
272 | ||
273 |
assert 'err' in result.json |
|
274 |
assert result.json['err'] |
|
275 |
assert 'Corrupted base64 content' in result.json['err_desc'] |
|
276 | ||
277 | ||
278 |
@patch('passerelle.utils.Request.get') |
|
279 |
@pytest.mark.parametrize( |
|
280 |
'status_code,body,expected_message', |
|
281 |
[ |
|
282 |
( |
|
283 |
400, |
|
284 |
'Le message d\'exception est \'Test error message\'. Pour plus d\'informations thanks a lot to use HTML as return of a json api.', |
|
285 |
'An error occured during the request to Signal Arrêtés: Test error message', |
|
286 |
), |
|
287 |
(500, 'Unmatched message', 'An error occured during the request to Signal Arrêtés'), |
|
288 |
(200, 'Invalid json', 'Expected valid json'), |
|
289 |
(200, '[]', 'Expected a dictionary with a GetCommunesResult key'), |
|
290 |
(200, '{}', 'Expected a dictionary with a GetCommunesResult key'), |
|
291 |
(200, '{"GetCommunesResult": "Invalid json"}', 'Expected valid json string at GetCommunesResult key'), |
|
292 |
(200, '{"GetCommunesResult": "{}"}', 'Expected a list'), |
|
293 |
], |
|
294 |
) |
|
295 |
def test_error_handling(mocked_get, app, connector, status_code, body, expected_message): |
|
296 |
endpoint = generic_endpoint_url('signal-arretes', 'cities', slug=connector.slug) |
|
297 |
mocked_get.return_value = response(status_code, body) |
|
298 |
result = app.get(endpoint) |
|
299 | ||
300 |
assert 'err' in result.json |
|
301 |
assert result.json['err'] |
|
302 |
assert expected_message in result.json['err_desc'] |
|
0 |
- |