# 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 . 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