330 lines
12 KiB
Python
330 lines
12 KiB
Python
# 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
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from httmock import HTTMock, response, urlmatch
|
|
from requests.exceptions import ConnectionError
|
|
|
|
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 'commentaire' not in data or data['commentaire'] == 'Wubba Lubba Dub Dub'
|
|
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 = b'Test file content'
|
|
|
|
|
|
@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetDocumentDemande/.*')
|
|
def mock_get_document_demande(url, request):
|
|
if url.path.endswith('unavailable'):
|
|
return response(
|
|
200,
|
|
json.dumps({'GetDocumentDemandeResult': json.dumps('fichier indisponible')}),
|
|
)
|
|
|
|
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',
|
|
'comment': '',
|
|
}
|
|
|
|
|
|
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',
|
|
'comment': 'Wubba Lubba Dub Dub',
|
|
}
|
|
)
|
|
|
|
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']
|
|
|
|
result = app.get(endpoint, params={'request_id': 'document_unavailable'})
|
|
|
|
assert 'err' in result.json
|
|
assert result.json['err']
|
|
assert 'fichier indisponible' 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']
|
|
|
|
|
|
@patch('passerelle.utils.Request.get')
|
|
def test_more_error_handling(mocked_get, app, connector):
|
|
endpoint = generic_endpoint_url('signal-arretes', 'cities', slug=connector.slug)
|
|
mocked_get.side_effect = ConnectionError
|
|
result = app.get(endpoint)
|
|
|
|
assert 'err' in result.json
|
|
assert result.json['err']
|
|
assert 'An error occured during the request to Signal Arrêtés' in result.json['err_desc']
|