187 lines
6.5 KiB
Python
187 lines
6.5 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# Copyright (C) 2021 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
|
|
import os
|
|
|
|
import httmock
|
|
import pytest
|
|
|
|
import tests.utils
|
|
from passerelle.contrib.sigerly.models import Sigerly
|
|
|
|
TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'sigerly')
|
|
|
|
|
|
@pytest.fixture
|
|
def connector(db):
|
|
return tests.utils.setup_access_rights(
|
|
Sigerly.objects.create(slug='test', base_url='https://sigerly.example.net')
|
|
)
|
|
|
|
|
|
def json_get_data(filename):
|
|
with open(os.path.join(TEST_BASE_DIR, "%s.json" % filename)) as fd:
|
|
return json.dumps(json.load(fd))
|
|
|
|
|
|
def get_endpoint(name):
|
|
return tests.utils.generic_endpoint_url('sigerly', name)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'status_code, content, err_desc',
|
|
[
|
|
(500, 'xxx', 'bad response code from API: 500'),
|
|
(400, 'xxx', 'bad response code from API: 400'),
|
|
(200, 'not json', 'invalid JSON content'),
|
|
],
|
|
)
|
|
def test_request_error(app, connector, status_code, content, err_desc):
|
|
endpoint = get_endpoint('query')
|
|
|
|
@httmock.all_requests
|
|
def sigerly_mock(url, request):
|
|
return httmock.response(status_code, content)
|
|
|
|
with httmock.HTTMock(sigerly_mock):
|
|
resp = app.post_json(endpoint, params={})
|
|
|
|
assert resp.json['err']
|
|
assert err_desc in resp.json['err_desc']
|
|
|
|
|
|
def test_create(app, connector):
|
|
endpoint = get_endpoint('create')
|
|
payload = {
|
|
'demandeur': 'Test webservice',
|
|
'id_typeinterv': '5',
|
|
'id_urgence': '1',
|
|
'id_qualification': '8',
|
|
'observations': 'Test webservice',
|
|
'elements': 'LIMW003D:LIMWW003C',
|
|
}
|
|
|
|
@httmock.urlmatch(netloc='sigerly.example.net', path='/createIntervention.php', method='POST')
|
|
def sigerly_mock(url, request):
|
|
assert request.headers['Accept'] == 'application/json'
|
|
assert json.loads(request.body)['elements'] == ['LIMW003D', 'LIMWW003C']
|
|
assert 'nom_rue' not in json.loads(request.body)
|
|
return httmock.response(200, json.dumps({'success': True, 'message': '7830'}))
|
|
|
|
with httmock.HTTMock(sigerly_mock):
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['message'] == '7830' # id unusable within query endpoint
|
|
|
|
|
|
def test_create_nom_rue(app, connector):
|
|
endpoint = get_endpoint('create')
|
|
payload = {
|
|
'demandeur': 'Test webservice',
|
|
'id_typeinterv': '5',
|
|
'id_urgence': '1',
|
|
'id_qualification': '8',
|
|
'observations': 'Test webservice',
|
|
'nom_rue': 'some place',
|
|
}
|
|
|
|
@httmock.urlmatch(netloc='sigerly.example.net', path='/createIntervention.php', method='POST')
|
|
def sigerly_mock(url, request):
|
|
assert request.headers['Accept'] == 'application/json'
|
|
assert json.loads(request.body)['nom_rue'] == 'some place'
|
|
assert 'elements' not in json.loads(request.body)
|
|
return httmock.response(200, json.dumps({'success': True, 'message': '7830'}))
|
|
|
|
with httmock.HTTMock(sigerly_mock):
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['message'] == '7830' # id unusable within query endpoint
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'success, message, err_desc',
|
|
[
|
|
(False, 'an error message', 'an error message'),
|
|
(True, '', 'No intervention id returned'),
|
|
],
|
|
)
|
|
def test_create_error(app, connector, success, message, err_desc):
|
|
endpoint = get_endpoint('create')
|
|
payload = {
|
|
'demandeur': 'Test webservice',
|
|
'id_typeinterv': '5',
|
|
'id_urgence': '1',
|
|
'id_qualification': '8',
|
|
'observations': 'Test webservice',
|
|
'elements': 'LIMW003D:LIMWW003C',
|
|
}
|
|
|
|
@httmock.urlmatch(netloc='sigerly.example.net', path='/createIntervention.php', method='POST')
|
|
def sigerly_mock(url, request):
|
|
assert request.headers['Accept'] == 'application/json'
|
|
return httmock.response(200, json.dumps({'success': success, 'message': message}))
|
|
|
|
with httmock.HTTMock(sigerly_mock):
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert resp.json['err']
|
|
assert resp.json['err_desc'] == err_desc
|
|
|
|
|
|
def test_query_id(app, connector):
|
|
endpoint = get_endpoint('query')
|
|
payload = {
|
|
'id_intervention': '10914',
|
|
}
|
|
|
|
@httmock.urlmatch(netloc='sigerly.example.net', path='/getIntervention.php', method='POST')
|
|
def sigerly_mock(url, request):
|
|
assert request.headers['Accept'] == 'application/json'
|
|
assert json.loads(request.body) == {'id_intervention': 10914}
|
|
return httmock.response(200, json_get_data('getIntervention_1'))
|
|
|
|
with httmock.HTTMock(sigerly_mock):
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert not resp.json['err']
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['id_demande_web'] == 10914
|
|
assert resp.json['data'][0]['idinterv'] == 22014
|
|
elements = resp.json['data'][0]['elements']
|
|
assert len(elements) == 2
|
|
assert [x['ident'] for x in elements] == ['LIMW003D', 'LIMW003C']
|
|
|
|
|
|
def test_query_filters(app, connector):
|
|
endpoint = get_endpoint('query')
|
|
payload = {
|
|
'date_debut_demande': '19/11/2020',
|
|
'date_fin_demande': '19/11/2020',
|
|
'insee': '::069291:::069283::',
|
|
}
|
|
|
|
@httmock.urlmatch(netloc='sigerly.example.net', path='/getIntervention.php', method='POST')
|
|
def sigerly_mock(url, request):
|
|
assert request.headers['Accept'] == 'application/json'
|
|
assert json.loads(request.body)['insee'] == ['069291', '069283']
|
|
return httmock.response(200, json_get_data('getIntervention_2'))
|
|
|
|
with httmock.HTTMock(sigerly_mock):
|
|
resp = app.post_json(endpoint, params=payload)
|
|
assert not resp.json['err']
|
|
assert len(resp.json['data']) == 2
|
|
assert [x['date_emission'] for x in resp.json['data']] == ['2020-11-19'] * 2
|
|
assert [x['inseecommune'] for x in resp.json['data']] == ['069291'] * 2
|