passerelle/tests/test_atal_rest.py

335 lines
12 KiB
Python

import base64
import json
import pytest
import responses
from django.contrib.contenttypes.models import ContentType
from passerelle.apps.atal_rest.models import AtalREST
from passerelle.base.models import AccessRight, ApiUser
@pytest.fixture()
def connector(db):
api = ApiUser.objects.create(username='all', keytype='', key='')
connector = AtalREST.objects.create(base_url='https://atal.invalid', slug='test', api_key='secret')
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
def test_check_status(connector):
with responses.RequestsMock() as rsps:
rsps.get(
'https://atal.invalid/api/Test',
status=200,
body=b'ok',
)
connector.check_status()
assert len(rsps.calls) == 1
assert rsps.calls[0].request.headers['X-API-Key'] == 'secret'
@pytest.mark.parametrize(
'geoloc',
[
{'latitude': '47.257117', 'longitude': '-0.07624'},
{'latitude': '47,257117', 'longitude': '-0,07624'},
{'latitude': 47.257117, 'longitude': -0.07624},
],
)
def test_worksrequest(app, connector, geoloc):
with responses.RequestsMock() as rsps:
rsps.post(
'https://atal.invalid/api/WorksRequests',
status=200,
json={'id': '1', 'RequestState': 0},
)
params = {
'desired_date': '2023-06-28',
'object': "test entr'ouvert Publik",
'recipient_id': '56',
'requester_id': '12',
'requesting_department_id': '17',
'localization': 'foo',
}
params.update(geoloc)
resp = app.post_json('/atal-rest/test/worksrequests', params=params)
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data']['id'] == '1'
assert json_resp['data']['RequestState'] == 0
assert json_resp['data']['RequestStateLabel'] == 'En attente'
request_data = json.loads(rsps.calls[0].request.body)
assert request_data['Localization'] == 'foo'
assert request_data['Latitude'] == 47.257117
assert request_data['Longitude'] == -0.07624
def test_worksrequest_status(app, connector):
with responses.RequestsMock() as rsps:
rsps.get(
'https://atal.invalid/api/WorksRequests/1',
status=200,
json={'status': 'foo', 'RequestState': 0},
)
resp = app.get('/atal-rest/test/worksrequest-status?worksrequests_id=1')
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data']['status'] == 'foo'
assert json_resp['data']['RequestState'] == 0
assert json_resp['data']['RequestStateLabel'] == 'En attente'
assert rsps.calls[0].request.params == {'$expand': 'Responses'}
def test_worksrequest_status_filter_responses(app, connector):
with responses.RequestsMock() as rsps:
rsps.get(
'https://atal.invalid/api/WorksRequests/1',
status=200,
json={
'status': 'foo',
'RequestState': 0,
'Responses': [
{
'ActionTypeId': 501,
'Description': 'Foo',
},
{
'ActionTypeId': 507,
'Description': 'Foo',
},
{
'ActionTypeId': 516,
'Description': 'Baz',
},
],
},
)
resp = app.get('/atal-rest/test/worksrequest-status?worksrequests_id=1&filter_responses=501,507')
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data']['status'] == 'foo'
assert json_resp['data']['RequestState'] == 0
assert json_resp['data']['RequestStateLabel'] == 'En attente'
assert json_resp['data']['Responses'] == [
{
'ActionTypeId': 501,
'Description': 'Foo',
},
{
'ActionTypeId': 507,
'Description': 'Foo',
},
]
assert rsps.calls[0].request.params == {'$expand': 'Responses'}
def test_worksrequest_intervention_status(app, connector):
with responses.RequestsMock() as rsps:
rsps.get(
'https://atal.invalid/api/WorksRequests/GetInterventionStates',
status=200,
json=[
{
'RequestId': 'cc8b7f6b-8ccf-4938-a648-09678feda679',
'InterventionState': 2,
'WorkState': 2,
'InterventionNumber': 'IN23090003',
}
],
)
resp = app.get('/atal-rest/test/worksrequest-intervention-status?number=DIT23070011')
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data']['InterventionState'] == 2
assert json_resp['data']['WorkState'] == 2
assert json_resp['data']['WorkStateLabel'] == 'En cours'
def test_worksrequest_intervention_status_empty_list(app, connector):
with responses.RequestsMock() as rsps:
rsps.get(
'https://atal.invalid/api/WorksRequests/GetInterventionStates',
status=200,
json=[],
)
resp = app.get('/atal-rest/test/worksrequest-intervention-status?number=DIT23070011')
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data'] == {'WorkStateLabel': ''}
def test_worksrequests_single_attachment(app, connector):
with responses.RequestsMock() as rsps:
rsps.post('https://atal.invalid/api/WorksRequests/1/Attachments', status=200, body=b'')
params = {'file': {'filename': 'bla', 'content': base64.b64encode(b'bla').decode('utf-8')}}
resp = app.post_json(
'/atal-rest/test/worksrequests-single-attachment?worksrequests_id=1', params=params
)
json_resp = resp.json
assert json_resp['err'] == 0
def test_worksrequests_single_attachment_no_data(app, connector):
with responses.RequestsMock() as rsps:
params = {
'file': '',
}
resp = app.post_json(
'/atal-rest/test/worksrequests-single-attachment?worksrequests_id=1', params=params
)
json_resp = resp.json
assert json_resp['err'] == 0
assert len(rsps.calls) == 0
def test_worksrequests_single_attachment_string_not_empty(app, connector):
params = {
'file': 'aaa',
}
app.post_json(
'/atal-rest/test/worksrequests-single-attachment?worksrequests_id=1', params=params, status=400
)
def test_worksrequests_single_attachment_error(app, connector):
with responses.RequestsMock() as rsps:
rsps.post(
'https://atal.invalid/api/WorksRequests/1/Attachments',
status=400,
json={
'type': 'https://tools.ietf.org/html/rfc7231#section-6.5.1',
'title': 'Bad Request',
'status': 400,
'"detail': 'No content","traceId":"00-1034a23a6cfbb7c508aa7e125a8e9a52-4570fc75745b7d1d-00',
},
)
params = {
'file': {'filename': 'bla', 'content': base64.b64encode(b'bla').decode('utf-8')},
}
resp = app.post_json(
'/atal-rest/test/worksrequests-single-attachment?worksrequests_id=1', params=params
)
json_resp = resp.json
assert json_resp['err'] == 1
assert json_resp['data']['title'] == 'Bad Request'
def test_worksrequests_attachments(app, connector):
with responses.RequestsMock() as rsps:
rsps.post('https://atal.invalid/api/WorksRequests/Attachments', status=200, body=b'')
params = {
'files': [
{'filename': 'bla', 'content': base64.b64encode(b'bla').decode('utf-8')},
{'filename': 'blo', 'content': base64.b64encode(b'blo').decode('utf-8')},
],
'worksrequests_ids': ['0', '1'],
}
resp = app.post_json('/atal-rest/test/worksrequests-attachments', params=params)
json_resp = resp.json
assert json_resp['err'] == 0
def test_worksrequests_attachments_no_data(app, connector):
with responses.RequestsMock() as rsps:
params = {
'files': ['', ''],
'worksrequests_ids': ['0', '1'],
}
resp = app.post_json('/atal-rest/test/worksrequests-attachments', params=params)
json_resp = resp.json
assert json_resp['err'] == 0
assert len(rsps.calls) == 0
def test_worksrequests_attachments_string_not_empty(app, connector):
params = {
'files': ['aa'],
'worksrequests_ids': ['0', '1'],
}
app.post_json('/atal-rest/test/worksrequests-attachments', params=params, status=400)
def test_worksrequests_attachments_error(app, connector):
with responses.RequestsMock() as rsps:
rsps.post(
'https://atal.invalid/api/WorksRequests/Attachments',
status=400,
json={
'type': 'https://tools.ietf.org/html/rfc7231#section-6.5.1',
'title': 'Bad Request',
'status': 400,
'"detail': 'No content","traceId":"00-1034a23a6cfbb7c508aa7e125a8e9a52-4570fc75745b7d1d-00',
},
)
params = {
'files': [
{'filename': 'bla', 'content': base64.b64encode(b'bla').decode('utf-8')},
{'filename': 'blo', 'content': base64.b64encode(b'blo').decode('utf-8')},
],
'worksrequests_ids': ['0', '1'],
}
resp = app.post_json('/atal-rest/test/worksrequests-attachments', params=params)
json_resp = resp.json
assert json_resp['err'] == 1
assert json_resp['data']['title'] == 'Bad Request'
def test_thirdparties_requesting_departments(app, connector):
with responses.RequestsMock() as rsps:
query_params = {'RequestType': '1001'}
rsps.get(
'https://atal.invalid/api/ThirdParties/RequestingDepartments',
match=[responses.matchers.query_param_matcher(query_params)],
status=200,
json=[
{
'Name': 'foo',
'Id': 1,
},
{
'Name': 'bar',
'Id': 2,
},
],
)
resp = app.get(
'/atal-rest/test/thirdparties-requesting-departments?request_type=1001',
)
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data'] == [
{'Name': 'foo', 'Id': 1, 'id': 1, 'text': 'foo'},
{'Name': 'bar', 'Id': 2, 'id': 2, 'text': 'bar'},
]
def test_users(app, connector):
with responses.RequestsMock() as rsps:
rsps.get(
'https://atal.invalid/api/Users',
status=200,
json=[
{
'Name': 'foo',
'Id': 1,
},
{
'Name': 'bar',
'Id': 2,
},
],
)
resp = app.get(
'/atal-rest/test/users',
)
json_resp = resp.json
assert json_resp['err'] == 0
assert json_resp['data'] == [
{'Name': 'foo', 'Id': 1, 'id': 1, 'text': 'foo'},
{'Name': 'bar', 'Id': 2, 'id': 2, 'text': 'bar'},
]