354 lines
12 KiB
Python
354 lines
12 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 base64
|
|
import json
|
|
|
|
import httmock
|
|
import pytest
|
|
from Cryptodome.Cipher import DES
|
|
from Cryptodome.Util.Padding import pad, unpad
|
|
from django.utils.encoding import force_bytes
|
|
|
|
import tests.utils
|
|
from passerelle.apps.esirius.models import ESirius
|
|
from passerelle.utils.jsonresponse import APIError
|
|
from tests.test_manager import login
|
|
|
|
CREATE_APPOINTMENT_PAYLOAD = {
|
|
'beginDate': '2021-02-24',
|
|
'beginTime': '16:40',
|
|
'endDate': '2021-02-24',
|
|
'endTime': '17:00',
|
|
'comment': 'commentaire',
|
|
'isoLanguage': 'fr',
|
|
'needsConfirmation': 'False',
|
|
'rdvChannel': 'WEBSERVICES',
|
|
'receptionChannel': 'WS',
|
|
'serviceId': '9',
|
|
'siteCode': 'site1',
|
|
'resources/id': '1',
|
|
'resources/key': '17',
|
|
'resources/type': 'STATION',
|
|
}
|
|
|
|
GET_APPOINTMENT_RESPONSE = '''
|
|
{
|
|
"beginDate" : "2021-02-26",
|
|
"beginTime" : "16:40",
|
|
"codeRDV" : "94PEP4",
|
|
"comment" : "coucou",
|
|
"endDate" : "2021-02-26",
|
|
"endTime" : "17:00",
|
|
"idSys" : 108840,
|
|
"isoLanguage" : "fr",
|
|
"motives" : [],
|
|
"needsConfirmation" : false,
|
|
"rdvChannel" : "EAPP0",
|
|
"receptionChannel" : "WS",
|
|
"resources" : {
|
|
"id" : 29,
|
|
"key" : "46",
|
|
"name" : "C1",
|
|
"type" : "STATION"
|
|
},
|
|
"serviceId" : "39",
|
|
"siteCode" : "site1",
|
|
"siteIdSys" : 5,
|
|
"user" : {
|
|
"additionalPersonalIdentity" : [],
|
|
"address" : {},
|
|
"civility" : "",
|
|
"idSys" : "95897"
|
|
}
|
|
}
|
|
'''
|
|
|
|
|
|
@pytest.fixture
|
|
def connector(db):
|
|
return tests.utils.setup_access_rights(
|
|
ESirius.objects.create(
|
|
slug='test', secret_id='xxx', secret_key='yyy', base_url='https://esirius.example.net'
|
|
)
|
|
)
|
|
|
|
|
|
def get_endpoint(name):
|
|
return tests.utils.generic_endpoint_url('esirius', name)
|
|
|
|
|
|
@pytest.mark.freeze_time('2021-01-26 15:13:6.880') # epoch + 1611673986.88 s
|
|
def test_token(connector):
|
|
connector.secret_id = 'eAppointment'
|
|
connector.secret_key = 'ES2I Info Caller Http Encryption Key'
|
|
connector.save()
|
|
|
|
@httmock.all_requests
|
|
def esirius_mock(url, request):
|
|
assert (
|
|
request.headers['token_info_caller']
|
|
== b'yM4zYAxT67Qvjd20riG3j0eu0t0Ku+HLlttj17Gul7zkruFaXX1J8BJ6sV2Ldgw40axfWh+ESAY='
|
|
)
|
|
return httmock.response(200)
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
connector.request('an/uri/', method='get', params="somes")
|
|
|
|
|
|
@pytest.mark.parametrize('secret_key', ['yyy', ''])
|
|
def test_pre_request(connector, secret_key):
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/an/uri/', method='GET')
|
|
def esirius_mock(url, request):
|
|
assert request.headers['Accept'] == 'application/json; charset=utf-8'
|
|
assert bool(request.headers.get('token_info_caller')) == bool(secret_key)
|
|
if secret_key:
|
|
des_key = pad(force_bytes(secret_key), 8)[:8]
|
|
cipher = DES.new(des_key, DES.MODE_ECB)
|
|
msg = cipher.decrypt(base64.b64decode(request.headers['token_info_caller']))
|
|
token = json.loads(unpad(msg, 8))
|
|
assert set(token) == {'caller', 'createInfo'}
|
|
assert token['caller'] == connector.secret_id
|
|
return httmock.response(200)
|
|
|
|
connector.secret_key = secret_key
|
|
connector.save()
|
|
with httmock.HTTMock(esirius_mock):
|
|
connector.request('an/uri/', method='get', params="somes")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'status_code, content, a_dict',
|
|
[
|
|
(400, '{"message": "help"}', {'message': 'help'}),
|
|
(500, 'not json', None),
|
|
],
|
|
)
|
|
def test_post_request(connector, status_code, content, a_dict):
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/an/uri/', method='GET')
|
|
def esirius_mock(url, request):
|
|
return httmock.response(status_code, content)
|
|
|
|
with pytest.raises(APIError) as exc:
|
|
with httmock.HTTMock(esirius_mock):
|
|
connector.request('an/uri/', params="somes")
|
|
|
|
assert exc.value.err
|
|
assert exc.value.data['status_code'] == status_code
|
|
assert exc.value.data['json_content'] == a_dict
|
|
|
|
|
|
@pytest.mark.parametrize('status_code, is_up', [(200, True), (500, False), (503, False)])
|
|
@pytest.mark.parametrize('content', ['wathever', '{"message": "help"}'])
|
|
def test_check_status(app, connector, status_code, is_up, content):
|
|
@httmock.all_requests
|
|
def esirius_mock(url, request):
|
|
return httmock.response(status_code, content)
|
|
|
|
if is_up:
|
|
with httmock.HTTMock(esirius_mock):
|
|
connector.check_status()
|
|
else:
|
|
with pytest.raises(APIError):
|
|
with httmock.HTTMock(esirius_mock):
|
|
connector.check_status()
|
|
|
|
|
|
def test_create_appointment(app, connector):
|
|
endpoint = get_endpoint('create-appointment')
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments/', method='POST')
|
|
def esirius_mock(url, request):
|
|
assert json.loads(request.body)['resources'] == {'id': '1', 'key': '17', 'type': 'STATION'}
|
|
return httmock.response(200, b'94PEP4')
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.post_json(endpoint, params=CREATE_APPOINTMENT_PAYLOAD)
|
|
|
|
assert not resp.json['err']
|
|
assert resp.json['data'] == {'id': '94PEP4', 'created': True}
|
|
|
|
|
|
def test_create_appointment_error_404(app, connector):
|
|
endpoint = get_endpoint('create-appointment')
|
|
|
|
# payload not providing or providing an unconfigured serviceId
|
|
payload = CREATE_APPOINTMENT_PAYLOAD
|
|
del payload['serviceId']
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments/', method='POST')
|
|
def esirius_mock(url, request):
|
|
return httmock.response(
|
|
404,
|
|
{
|
|
'code': 'Not Found',
|
|
'type': 'com.es2i.planning.api.exception.NoService4RDVException',
|
|
'message': "Le rendez-vous {0} n'a pas créé",
|
|
},
|
|
)
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.post_json(endpoint, params=payload)
|
|
|
|
assert resp.json['err']
|
|
assert resp.json['data']['status_code'] == 404
|
|
assert resp.json['data']['json_content'] == {
|
|
'code': 'Not Found',
|
|
'type': 'com.es2i.planning.api.exception.NoService4RDVException',
|
|
'message': "Le rendez-vous {0} n'a pas créé",
|
|
}
|
|
|
|
|
|
def test_create_appointment_error_500(app, connector):
|
|
endpoint = get_endpoint('create-appointment')
|
|
|
|
# payload not providing beginTime
|
|
payload = {'beginDate': '2021-02-23'}
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments/', method='POST')
|
|
def esirius_mock(url, request):
|
|
return httmock.response(500, 'java stack')
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.post_json(endpoint, params=payload)
|
|
|
|
assert resp.json['err']
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert resp.json['err_desc'] == "error status:500 None, content:'java stack'"
|
|
assert resp.json['data']['status_code'] == 500
|
|
assert resp.json['data']['json_content'] is None
|
|
|
|
|
|
def test_update_appointment(app, connector):
|
|
endpoint = get_endpoint('update-appointment')
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments', method='PUT')
|
|
def esirius_mock(url, request):
|
|
assert 'codeRDV' in json.loads(request.body).keys()
|
|
return httmock.response(200, b'')
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.post_json(endpoint + '?id=94PEP4', params=CREATE_APPOINTMENT_PAYLOAD)
|
|
|
|
assert not resp.json['err']
|
|
assert resp.json['data'] == {'id': '94PEP4', 'updated': True}
|
|
|
|
|
|
def test_update_appointment_error(app, connector):
|
|
endpoint = get_endpoint('update-appointment')
|
|
payload = CREATE_APPOINTMENT_PAYLOAD
|
|
payload['idSys'] = 42
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments', method='PUT')
|
|
def esirius_mock(url, request):
|
|
raise ResourceWarning
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.post_json(endpoint + '?id=94PEP4', params=payload, status=400)
|
|
|
|
assert resp.json['err']
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert resp.json['err_desc'] == "idSys: 42 is not of type 'string'"
|
|
|
|
|
|
def test_get_appointment(app, connector):
|
|
endpoint = get_endpoint('get-appointment')
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments/94PEP4/', method='GET')
|
|
def esirius_mock(url, request):
|
|
return httmock.response(200, GET_APPOINTMENT_RESPONSE)
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.get(endpoint + '?id=94PEP4')
|
|
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['codeRDV'] == '94PEP4'
|
|
assert resp.json['data'] == json.loads(GET_APPOINTMENT_RESPONSE)
|
|
|
|
|
|
def test_get_appointment_error(app, connector):
|
|
endpoint = get_endpoint('get-appointment')
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments/42OUPS/', method='GET')
|
|
def esirius_mock(url, request):
|
|
return httmock.response(404, '{"code":"Not Found", "message":"Le rendez-vous {0} n\'existe pas"}')
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.get(endpoint + '?id=42OUPS')
|
|
|
|
assert resp.json['err']
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert resp.json['data']['status_code'] == 404
|
|
assert resp.json['data']['json_content'] == {
|
|
"code": "Not Found",
|
|
"message": "Le rendez-vous {0} n'existe pas",
|
|
}
|
|
|
|
|
|
def test_delete_appointment(app, connector):
|
|
endpoint = get_endpoint('delete-appointment')
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments/94PEP4/', method='DELETE')
|
|
def esirius_mock(url, request):
|
|
return httmock.response(200, b'')
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.delete(endpoint + '?id=94PEP4')
|
|
|
|
assert not resp.json['err']
|
|
assert resp.json['data'] == {'id': '94PEP4', 'deleted': True}
|
|
|
|
|
|
def test_delete_appointment_error(app, connector):
|
|
endpoint = get_endpoint('delete-appointment')
|
|
|
|
@httmock.urlmatch(netloc='esirius.example.net', path='/appointments/42OUPS/', method='DELETE')
|
|
def esirius_mock(url, request):
|
|
return httmock.response(304, b'')
|
|
|
|
with httmock.HTTMock(esirius_mock):
|
|
resp = app.delete(endpoint + '?id=42OUPS')
|
|
|
|
assert resp.json['err']
|
|
assert resp.json['err_desc'] == 'Appointment not found'
|
|
|
|
|
|
def test_manager(db, app, admin_user, connector):
|
|
url = '/%s/%s/' % (connector.get_connector_slug(), connector.slug)
|
|
connector.title = 'Test'
|
|
connector.description = 'Test eSirius'
|
|
connector.save()
|
|
login(app)
|
|
resp = app.get(url)
|
|
resp = resp.click('Edit')
|
|
assert resp.html.find('div', {'id': 'id_secret_key_p'}).input['value'] == 'yyy'
|
|
resp = resp.form.submit()
|
|
assert (
|
|
'DES key must be 8 bytes long'
|
|
in resp.html.find('div', {'id': 'id_secret_key_p'}).find('div', {'class': 'error'}).text
|
|
)
|
|
resp.form['secret_key'] = '8 bytes!'
|
|
resp = resp.form.submit()
|
|
assert ESirius.objects.get().secret_key == '8 bytes!'
|
|
|
|
# accept an empty key
|
|
resp = app.get(url)
|
|
resp = resp.click('Edit')
|
|
assert resp.html.find('div', {'id': 'id_secret_key_p'}).input['value'] == '8 bytes!'
|
|
resp.form['secret_key'] = ''
|
|
resp = resp.form.submit()
|
|
assert ESirius.objects.get().secret_key == ''
|