276 lines
8.2 KiB
Python
276 lines
8.2 KiB
Python
import base64
|
|
import json
|
|
|
|
import pytest
|
|
import responses
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
import tests.utils
|
|
from passerelle.apps.esabora.models import Esabora
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
|
|
ESABORA_DO_SEARCH_RESPONSE = {
|
|
'searchId': '23568',
|
|
'nbResults': 2,
|
|
'columnList': [
|
|
'Column 1',
|
|
'Column 2',
|
|
'Column 3',
|
|
],
|
|
'keyList': ['internal.id'],
|
|
'rowList': [
|
|
{
|
|
'columnDataList': [
|
|
'Foo 1',
|
|
'Foo 2',
|
|
'Foo 3',
|
|
],
|
|
'keyDataList': ['id1'],
|
|
},
|
|
{
|
|
'columnDataList': [
|
|
'Bar 1',
|
|
'Bar 2',
|
|
'Bar 3',
|
|
],
|
|
'keyDataList': ['id2'],
|
|
},
|
|
],
|
|
}
|
|
ESABORA_DO_TREATMENT_RESPONSE = {
|
|
'action': 'insert',
|
|
'keyList': ['internal.id'],
|
|
'keyDataList': ['14'],
|
|
}
|
|
|
|
|
|
@pytest.fixture()
|
|
def connector(db):
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
connector = Esabora.objects.create(
|
|
service_url='http://example.esabora/ws/rest/', api_key='1234', slug='test'
|
|
)
|
|
obj_type = ContentType.objects.get_for_model(connector)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=connector.pk
|
|
)
|
|
return connector
|
|
|
|
|
|
@responses.activate
|
|
def test_authentication(app, connector):
|
|
url = tests.utils.generic_endpoint_url('esabora', 'do-search')
|
|
responses.add(
|
|
responses.POST,
|
|
f'{connector.service_url}mult/',
|
|
json=ESABORA_DO_SEARCH_RESPONSE,
|
|
status=200,
|
|
)
|
|
|
|
app.post_json(url, params={'search_name': 'foo', 'criterions/bar': 'noop'})
|
|
assert len(responses.calls) == 1
|
|
assert responses.calls[0].request.headers['authorization'] == f'Bearer {connector.api_key}'
|
|
|
|
|
|
@responses.activate
|
|
def test_do_treatment(app, connector):
|
|
url = tests.utils.generic_endpoint_url('esabora', 'do-treatment')
|
|
responses.add(
|
|
responses.POST,
|
|
f'{connector.service_url}modbdd/',
|
|
json=ESABORA_DO_TREATMENT_RESPONSE,
|
|
status=200,
|
|
)
|
|
|
|
file_content1 = b'this is a test file'
|
|
file_content2 = b'this is another test file'
|
|
file_content3 = b'this is yet another test file'
|
|
|
|
payload = {
|
|
'treatment_name': 'Import HISTOLOGE',
|
|
'Adresse_Latitude': 12.3,
|
|
'Adresse_Ville': 'Marseille',
|
|
# documents that will be unflattened
|
|
'PJ_Documents/0': {
|
|
'filename': 'test1.pdf',
|
|
'content_type': 'application/pdf',
|
|
'content': base64.b64encode(file_content1).decode(),
|
|
},
|
|
'PJ_Documents/1': {
|
|
'filename': 'test2.pdf',
|
|
'content_type': 'application/pdf',
|
|
'content': base64.b64encode(file_content2).decode(),
|
|
},
|
|
# empty field, will be skipped
|
|
'PJ_Documents/2': '',
|
|
# ensure we handle single documents as well
|
|
'PJ_Documents_Autre': {
|
|
'filename': 'test3.pdf',
|
|
'content_type': 'application/pdf',
|
|
'content': base64.b64encode(file_content3).decode(),
|
|
},
|
|
}
|
|
|
|
expected_payload = {
|
|
'treatmentName': 'Import HISTOLOGE',
|
|
'fieldList': [
|
|
{'fieldName': 'Adresse_Latitude', 'fieldValue': 12.3},
|
|
{'fieldName': 'Adresse_Ville', 'fieldValue': 'Marseille'},
|
|
{
|
|
'fieldName': 'PJ_Documents',
|
|
'fieldDocumentUpdate': 1,
|
|
'fieldValue': [
|
|
{
|
|
'documentName': 'test1.pdf',
|
|
'documentContent': base64.b64encode(file_content1).decode(),
|
|
# len(file_content1)
|
|
'documentSize': 19,
|
|
},
|
|
{
|
|
'documentName': 'test2.pdf',
|
|
'documentContent': base64.b64encode(file_content2).decode(),
|
|
'documentSize': 25,
|
|
},
|
|
],
|
|
},
|
|
{
|
|
'fieldName': 'PJ_Documents_Autre',
|
|
'fieldDocumentUpdate': 1,
|
|
'fieldValue': [
|
|
{
|
|
'documentName': 'test3.pdf',
|
|
'documentContent': base64.b64encode(file_content3).decode(),
|
|
'documentSize': 29,
|
|
}
|
|
],
|
|
},
|
|
],
|
|
}
|
|
|
|
response = app.post_json(url, params=payload)
|
|
assert response.json == {'err': 0, 'action': 'insert', 'id': '14', 'internalid': '14', 'text': '14'}
|
|
|
|
assert len(responses.calls) == 1
|
|
assert responses.calls[0].request.params['task'] == 'doTreatment'
|
|
response_data = json.loads(responses.calls[0].request.body)
|
|
assert response_data == expected_payload
|
|
|
|
|
|
@responses.activate
|
|
def test_do_search(app, connector):
|
|
url = tests.utils.generic_endpoint_url('esabora', 'do-search')
|
|
responses.add(
|
|
responses.POST,
|
|
f'{connector.service_url}mult/',
|
|
json=ESABORA_DO_SEARCH_RESPONSE,
|
|
status=200,
|
|
)
|
|
|
|
payload = {'search_name': 'WS_ETAT_SAS', 'criterions/foo': 'bar'}
|
|
|
|
expected_payload = {
|
|
'searchName': 'WS_ETAT_SAS',
|
|
'criterionList': [{'criterionName': 'foo', 'criterionValueList': ['bar']}],
|
|
}
|
|
|
|
response = app.post_json(url, params=payload)
|
|
|
|
expected = {
|
|
'err': 0,
|
|
'data': [
|
|
{
|
|
'id': 'id1',
|
|
'text': 'Foo 1',
|
|
'internalid': 'id1',
|
|
'column_1': 'Foo 1',
|
|
'column_2': 'Foo 2',
|
|
'column_3': 'Foo 3',
|
|
},
|
|
{
|
|
'id': 'id2',
|
|
'text': 'Bar 1',
|
|
'internalid': 'id2',
|
|
'column_1': 'Bar 1',
|
|
'column_2': 'Bar 2',
|
|
'column_3': 'Bar 3',
|
|
},
|
|
],
|
|
'meta': {
|
|
'searchId': '23568',
|
|
'nbResults': 2,
|
|
'columns_name': {
|
|
'column_1': 'Column 1',
|
|
'column_2': 'Column 2',
|
|
'column_3': 'Column 3',
|
|
},
|
|
'keys_name': {
|
|
'internalid': 'internal.id',
|
|
},
|
|
},
|
|
}
|
|
assert response.json == expected
|
|
assert responses.calls[0].request.params['task'] == 'doSearch'
|
|
assert json.loads(responses.calls[0].request.body) == expected_payload
|
|
|
|
|
|
@responses.activate
|
|
def test_do_treatment_arbitrary_endpoint(app, connector):
|
|
url = tests.utils.generic_endpoint_url('esabora', 'do-treatment')
|
|
responses.add(
|
|
responses.POST,
|
|
f'{connector.service_url}addevt/',
|
|
json=ESABORA_DO_TREATMENT_RESPONSE,
|
|
status=200,
|
|
)
|
|
|
|
payload = {
|
|
'endpoint': 'addevt',
|
|
'treatment_name': 'Import Event',
|
|
'Adresse_Latitude': 12.3,
|
|
'Adresse_Ville': 'Marseille',
|
|
}
|
|
|
|
expected_payload = {
|
|
'treatmentName': 'Import Event',
|
|
'fieldList': [
|
|
{'fieldName': 'Adresse_Latitude', 'fieldValue': 12.3},
|
|
{'fieldName': 'Adresse_Ville', 'fieldValue': 'Marseille'},
|
|
],
|
|
}
|
|
|
|
response = app.post_json(url, params=payload)
|
|
assert response.json == {'err': 0, 'action': 'insert', 'id': '14', 'internalid': '14', 'text': '14'}
|
|
|
|
assert len(responses.calls) == 1
|
|
assert responses.calls[0].request.params['task'] == 'doTreatment'
|
|
response_data = json.loads(responses.calls[0].request.body)
|
|
assert response_data == expected_payload
|
|
|
|
|
|
@responses.activate
|
|
def test_post_raises_proper_error(app, connector):
|
|
url = tests.utils.generic_endpoint_url('esabora', 'do-treatment')
|
|
responses.add(
|
|
responses.POST,
|
|
f'{connector.service_url}addevt/',
|
|
json={'foo': 'bar'},
|
|
status=400,
|
|
)
|
|
|
|
payload = {
|
|
'endpoint': 'addevt',
|
|
'treatment_name': 'Import Event',
|
|
'Adresse_Latitude': 12.3,
|
|
'Adresse_Ville': 'Marseille',
|
|
}
|
|
|
|
expected_response = {
|
|
'err': 1,
|
|
'err_class': 'passerelle.utils.jsonresponse.APIError',
|
|
'data': {'content': {'foo': 'bar'}, 'status_code': 400},
|
|
'err_desc': 'Esabora platform "http://example.esabora/ws/rest/" answered with HTTP error',
|
|
}
|
|
|
|
response = app.post_json(url, params=payload)
|
|
assert response.json == expected_response
|