428 lines
18 KiB
Python
428 lines
18 KiB
Python
# Passerelle - uniform access to data 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; exclude 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.deepcopy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from django.urls import reverse
|
|
from requests import Request
|
|
from requests.exceptions import ConnectionError
|
|
|
|
import tests.utils
|
|
from passerelle.apps.astech.models import ASTech
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(db):
|
|
return tests.utils.setup_access_rights(
|
|
ASTech.objects.create(
|
|
slug='test',
|
|
base_url='https://astech.example.net/app.php/',
|
|
basic_auth_username='ENTROUV',
|
|
basic_auth_password='password',
|
|
)
|
|
)
|
|
|
|
|
|
CONNECTION_RESPONSE = """
|
|
{"default":"TEST","connections":{"TEST":{"name":"TEST","code":"TEST","clientId":"1234","secret":"9876"}}}
|
|
"""
|
|
|
|
AUTH_RESPONSE = """
|
|
{"access_token":"4242","expires_in":604800,"token_type":"bearer","scope":null,"refresh_token":"4343","database_version":"6.06.210","gamme_nomade":null,"connection_id":"TEST","connection_name":"TEST","user_detail":"Demande Publik","mercure_token":"eyJ0-BsWy","mercure_topics":null,"mercure_base_url":"/.well-known/mercure","module_options":null,"astech_serveur":"6.06.218.00.00"}
|
|
"""
|
|
|
|
SERVICES_RESPONSE = """
|
|
{"ABC":"ABC / AH BE CE","123":"123 / FOOBAR","XFO":"XFO / BARFOO"}
|
|
"""
|
|
|
|
COMPANY_RESPONSE = '"99"'
|
|
|
|
COMPANIES_RESPONSE = """
|
|
{"01":"01 / SERVICES TECHNIQUES","10":"10 / DIRECTION BATIMENT","11":"11 / PLOMBERIE"}
|
|
"""
|
|
|
|
LABELS_RESPONSE = """
|
|
{"1":"1 / CHANGEMENT AMPOULE","2":"2 / FUITE","3":"3 / SERRURE CASSEE","4":"4 / WC BOUCHE"}
|
|
"""
|
|
|
|
PARAMETER_RESPONSE = '{"LIBELDEMDEF":"O"}'
|
|
|
|
CREATE_DEMAND_RESPONSE = """
|
|
{"sgesdemNum": "000000000001234"}
|
|
"""
|
|
|
|
ADD_DOCUMENT_RESPONSE = """
|
|
""
|
|
"""
|
|
|
|
POSITION_RESPONSE = """
|
|
{"position":"E","positionLib":"Envoi atelier","info":null}
|
|
"""
|
|
|
|
POSITIONS_RESPONSE = """
|
|
[{"position":"A","positionLib":"En attente","color":"0, 0, 0"},{"position":"E","positionLib":"Envoi","color":"190, 190, 0"},{"position":"C","positionLib":"En cours","color":"255, 0, 0"},{"position":"D","positionLib":"Envoi signataire","color":"255, 255, 113"},{"position":"T","positionLib":"Termin\u00e9","color":"0, 0, 0"},{"position":"I","positionLib":"\u00c9dition devis","color":"0, 255, 255"},{"position":"R","positionLib":"Refus","color":"255, 0, 0"},{"position":"V","positionLib":"V\u00e9rification","color":"0, 255, 0"},{"position":"F","positionLib":"Devis effectu\u00e9","color":"153, 204, 255"},{"position":"P","positionLib":"Livraison partielle","color":"255, 102, 0"},{"position":"L","positionLib":"Livraison","color":"128, 0, 0"}]
|
|
"""
|
|
|
|
|
|
@mock.patch("passerelle.utils.Request.request")
|
|
def test_connections(mocked_request, app, setup):
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=CONNECTION_RESPONSE, status_code=200)
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "connections"},
|
|
)
|
|
response = app.get(endpoint)
|
|
assert mocked_request.call_args[0][0] == 'get'
|
|
assert mocked_request.call_args[0][1].endswith("connection/all")
|
|
assert mocked_request.call_count == 1
|
|
assert response.json["err"] == 0
|
|
assert response.json["data"]["default"] == "TEST"
|
|
|
|
ASTech.objects.update(connection="OTHER")
|
|
response = app.get(endpoint)
|
|
assert response.json["err"] == 1
|
|
|
|
ASTech.objects.update(connection="TEST")
|
|
response = app.get(endpoint)
|
|
assert response.json["err"] == 0
|
|
assert response.json["data"]["default"] == "TEST"
|
|
|
|
# bad response
|
|
mocked_request.return_value = tests.utils.FakedResponse(
|
|
content='{"msg": "not found"}', status_code=404, reason="Not Found"
|
|
)
|
|
response = app.get(endpoint)
|
|
assert response.json["err"] == 1
|
|
assert response.json["err_class"].endswith("APIError")
|
|
assert response.json["err_desc"] == "AS-TECH response: 404 Not Found"
|
|
assert response.json["data"]["error"]["status"] == 404
|
|
assert response.json["data"]["error"]["content"]["msg"] == "not found"
|
|
mocked_request.return_value = tests.utils.FakedResponse(
|
|
content="crash", status_code=500, reason="Crashhhh"
|
|
)
|
|
response = app.get(endpoint)
|
|
assert response.json["err"] == 1
|
|
assert response.json["err_class"].endswith("APIError")
|
|
assert response.json["err_desc"] == "AS-TECH response: 500 Crashhhh"
|
|
assert response.json["data"]["error"]["status"] == 500
|
|
assert response.json["data"]["error"]["content"] == "crash"
|
|
mocked_request.return_value = tests.utils.FakedResponse(content="not json", status_code=200, reason="OK")
|
|
response = app.get(endpoint)
|
|
assert response.json["err"] == 1
|
|
assert response.json["err_class"].endswith("APIError")
|
|
assert response.json["err_desc"].startswith("invalid JSON in response:")
|
|
mocked_request.side_effect = ConnectionError('mocked error', request=Request())
|
|
response = app.get(endpoint)
|
|
assert response.json["err"] == 1
|
|
assert response.json["err_class"].endswith("APIError")
|
|
assert response.json["err_desc"] == 'connection error: mocked error'
|
|
|
|
|
|
@mock.patch("passerelle.utils.Request.request")
|
|
def test_authorization(mocked_request, app, setup):
|
|
mocked_request.side_effect = [
|
|
tests.utils.FakedResponse(content=CONNECTION_RESPONSE, status_code=200),
|
|
tests.utils.FakedResponse(content=AUTH_RESPONSE, status_code=200),
|
|
]
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "authorization"},
|
|
)
|
|
response = app.get(endpoint)
|
|
assert mocked_request.call_count == 2
|
|
assert mocked_request.call_args_list[0][0][0] == 'get'
|
|
assert mocked_request.call_args_list[0][0][1].endswith('connection/all') # get client_id
|
|
assert mocked_request.call_args_list[1][0][0] == 'post'
|
|
assert mocked_request.call_args_list[1][0][1].endswith('oauth/v2/auth') # get access_token
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['access_token'] == '4242'
|
|
assert response.json['data']['connection_id'] == 'TEST'
|
|
|
|
# test cache
|
|
response = app.get(endpoint)
|
|
assert mocked_request.call_count == 2
|
|
|
|
|
|
@mock.patch("passerelle.utils.Request.request")
|
|
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
|
def test_services(mocked_auth, mocked_request, app, setup):
|
|
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "services"},
|
|
)
|
|
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=SERVICES_RESPONSE, status_code=200)
|
|
response = app.get(endpoint)
|
|
assert mocked_request.call_count == 1
|
|
assert mocked_request.call_args[0][0] == 'post'
|
|
assert mocked_request.call_args[0][1].endswith('/sousservices/invoke')
|
|
assert mocked_request.call_args[1]['params']['access_token'] == '4242'
|
|
assert mocked_request.call_args[1]['params']['connection_id'] == 'TEST'
|
|
assert response.json['err'] == 0
|
|
assert response.json['data'] == [
|
|
{"id": "123", "text": "123 / FOOBAR"},
|
|
{"id": "ABC", "text": "ABC / AH BE CE"},
|
|
{"id": "XFO", "text": "XFO / BARFOO"},
|
|
]
|
|
|
|
|
|
@mock.patch("passerelle.utils.Request.request")
|
|
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
|
def test_companies(mocked_auth, mocked_request, app, setup):
|
|
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "companies"},
|
|
)
|
|
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=COMPANIES_RESPONSE, status_code=200)
|
|
response = app.get(endpoint)
|
|
assert mocked_request.call_count == 1
|
|
assert mocked_request.call_args[0][0] == 'post'
|
|
assert mocked_request.call_args[0][1].endswith('/societes_demandeur/invoke')
|
|
assert mocked_request.call_args[1]['json']['codeDemandeur'] == 'ENTROUV'
|
|
assert response.json['err'] == 0
|
|
assert response.json['data'] == [
|
|
{"id": "01", "text": "01 / SERVICES TECHNIQUES"},
|
|
{"id": "10", "text": "10 / DIRECTION BATIMENT"},
|
|
{"id": "11", "text": "11 / PLOMBERIE"},
|
|
]
|
|
|
|
|
|
@mock.patch("passerelle.utils.Request.request")
|
|
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
|
def test_labels(mocked_auth, mocked_request, app, setup):
|
|
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "labels"},
|
|
)
|
|
|
|
mocked_request.side_effect = [
|
|
tests.utils.FakedResponse(content=COMPANY_RESPONSE, status_code=200),
|
|
tests.utils.FakedResponse(content=LABELS_RESPONSE, status_code=200),
|
|
]
|
|
response = app.get(endpoint, status=200)
|
|
assert mocked_request.call_count == 2
|
|
assert mocked_request.call_args_list[0][0][0] == 'post'
|
|
assert mocked_request.call_args_list[0][0][1].endswith('societe_demandeur/invoke') # get company (99)
|
|
assert mocked_request.call_args_list[1][0][0] == 'post'
|
|
assert mocked_request.call_args_list[1][0][1].endswith('libelles_predefinis/invoke') # get labels
|
|
assert mocked_request.call_args_list[1][1]['json'] == {'societeDemandeur': '99'}
|
|
assert response.json['err'] == 0
|
|
assert response.json['data'] == [
|
|
{"id": "1", "text": "1 / CHANGEMENT AMPOULE"},
|
|
{"id": "2", "text": "2 / FUITE"},
|
|
{"id": "3", "text": "3 / SERRURE CASSEE"},
|
|
{"id": "4", "text": "4 / WC BOUCHE"},
|
|
]
|
|
|
|
mocked_request.reset_mock(side_effect=True)
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=LABELS_RESPONSE, status_code=200)
|
|
response = app.get(endpoint + '?company=42', status=200)
|
|
assert mocked_request.call_count == 1
|
|
assert mocked_request.call_args[0][0] == 'post'
|
|
assert mocked_request.call_args[0][1].endswith('libelles_predefinis/invoke') # get labels
|
|
assert mocked_request.call_args[1]['json'] == {'societeDemandeur': '42'}
|
|
assert response.json['err'] == 0
|
|
assert len(response.json['data']) == 4
|
|
|
|
|
|
@mock.patch("passerelle.utils.Request.request")
|
|
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
|
def test_parameter(mocked_auth, mocked_request, app, setup):
|
|
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "parameter"},
|
|
)
|
|
|
|
mocked_request.side_effect = [
|
|
tests.utils.FakedResponse(content=COMPANY_RESPONSE, status_code=200),
|
|
tests.utils.FakedResponse(content=PARAMETER_RESPONSE, status_code=200),
|
|
]
|
|
response = app.get(endpoint + '?name=LIBELDEMDEF', status=200)
|
|
assert mocked_request.call_count == 2
|
|
assert mocked_request.call_args_list[0][0][0] == 'post'
|
|
assert mocked_request.call_args_list[0][0][1].endswith('societe_demandeur/invoke') # get company (99)
|
|
assert mocked_request.call_args_list[1][0][0] == 'get'
|
|
assert mocked_request.call_args_list[1][0][1].endswith('/common/getparam/LIBELDEMDEF/99') # get param
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['LIBELDEMDEF'] == 'O'
|
|
|
|
mocked_request.reset_mock(side_effect=True)
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=PARAMETER_RESPONSE, status_code=200)
|
|
response = app.get(endpoint + '?name=LIBELDEMDEF&company=00', status=200)
|
|
assert mocked_request.call_count == 1
|
|
assert mocked_request.call_args[0][0] == 'get'
|
|
assert mocked_request.call_args[0][1].endswith('/common/getparam/LIBELDEMDEF/00') # get param
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['LIBELDEMDEF'] == 'O'
|
|
|
|
response = app.get(endpoint, status=400)
|
|
|
|
|
|
@mock.patch("passerelle.utils.Request.request")
|
|
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
|
def test_create_demand(mocked_auth, mocked_request, app, setup):
|
|
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "create-demand"},
|
|
)
|
|
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=CREATE_DEMAND_RESPONSE, status_code=201)
|
|
demand = {
|
|
'company': '99',
|
|
'service': '11',
|
|
'label': '4',
|
|
'subject': '4 / WC BOUCHE',
|
|
'name': 'Super Mario',
|
|
'description': 'Come with Luigi please',
|
|
'email': 'peach@example.net',
|
|
'phone1': '123',
|
|
'phone2': '321',
|
|
'address1': 'Nintendo Tokyo Branch Office',
|
|
'address2': 'Chiyoda-ku',
|
|
'address3': 'Tokyo 101-0054',
|
|
}
|
|
response = app.post_json(endpoint, params=demand, status=200)
|
|
assert mocked_request.call_count == 1
|
|
assert mocked_request.call_args[0][0] == 'post'
|
|
assert mocked_request.call_args[0][1].endswith('interface-citoyenne/demande-intervention')
|
|
assert mocked_request.call_args[1]['json'] == {
|
|
'interface_citoyenne_demande': {
|
|
'sgesdemSoc': '99',
|
|
'sgesdemSserv': '11',
|
|
'sgesdemLibdef': '4',
|
|
'sgesdemNdt': '4 / WC BOUCHE',
|
|
'sgesdemLibelle': 'Come with Luigi please',
|
|
'sgesdemCplnom': 'Super Mario',
|
|
'sgesdemCplemail': 'peach@example.net',
|
|
'sgesdemCpltel1': '123',
|
|
'sgesdemCpltel2': '321',
|
|
'sgesdemCpladr1': 'Nintendo Tokyo Branch Office',
|
|
'sgesdemCpladr2': 'Chiyoda-ku',
|
|
'sgesdemCpladr3': 'Tokyo 101-0054',
|
|
}
|
|
}
|
|
assert response.json['err'] == 0
|
|
assert response.json['data']['demand_id'] == '000000000001234'
|
|
|
|
# mock invalid AS-TECH response
|
|
mocked_request.return_value = tests.utils.FakedResponse(content='{"foo":"bar"}', status_code=200)
|
|
response = app.post_json(endpoint, params=demand, status=200)
|
|
assert mocked_request.call_count == 2
|
|
assert response.json['err'] == 1
|
|
assert response.json['err_desc'].startswith('no sgesdemNum in response: ')
|
|
|
|
# send AsTech named params
|
|
demand['sgesdemArbo'] = 1
|
|
demand['sgesdemAff'] = 'E'
|
|
response = app.post_json(endpoint, params=demand, status=200)
|
|
assert mocked_request.call_args[1]['json']['interface_citoyenne_demande']['sgesdemArbo'] == 1
|
|
assert mocked_request.call_args[1]['json']['interface_citoyenne_demande']['sgesdemAff'] == 'E'
|
|
|
|
# test invalid requests
|
|
response = app.get(endpoint, status=405)
|
|
response = app.post_json(endpoint, status=400)
|
|
response = app.post_json(endpoint, params={'foo': 'bar'}, status=400)
|
|
|
|
# add a document
|
|
mocked_request.reset_mock()
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "add-document"},
|
|
)
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=ADD_DOCUMENT_RESPONSE, status_code=201)
|
|
document = {
|
|
'demand_id': '000000000001234',
|
|
'title': 'test document',
|
|
'document': {
|
|
'filename': 'test.txt',
|
|
'content_type': 'text/plain',
|
|
'content': 'Zm9vCg==', # base64(foo)
|
|
},
|
|
}
|
|
response = app.post_json(endpoint, params=document, status=200)
|
|
assert mocked_request.call_count == 1
|
|
assert mocked_request.call_args[0][0] == 'post'
|
|
assert mocked_request.call_args[0][1].endswith('document/sgesdemNum/000000000001234')
|
|
assert mocked_request.call_args[1]['params']['docTitre'] == 'test document'
|
|
assert mocked_request.call_args[1]['params']['docFile'] == 'test.txt'
|
|
assert mocked_request.call_args[1]['files'] == {'file0': ('test.txt', b'foo\n', 'text/plain')}
|
|
assert response.json['err'] == 0
|
|
assert response.json['data'] == ""
|
|
|
|
|
|
@mock.patch("passerelle.utils.Request.request")
|
|
@mock.patch("passerelle.apps.astech.models.ASTech.get_authorization")
|
|
def test_positions(mocked_auth, mocked_request, app, setup):
|
|
mocked_auth.return_value = {'access_token': '4242', 'connection_id': 'TEST'}
|
|
|
|
# position of a demand
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "demand-position"},
|
|
)
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=POSITION_RESPONSE, status_code=200)
|
|
response = app.get(endpoint + '?demand_id=000000000001234', status=200)
|
|
assert mocked_request.call_count == 1
|
|
assert mocked_request.call_args[0][0] == 'get'
|
|
assert mocked_request.call_args[0][1].endswith('apicli/demande/position/000000000001234')
|
|
assert response.json['err'] == 0
|
|
assert response.json['data'] == {
|
|
"position": "E",
|
|
"positionLib": "Envoi atelier",
|
|
"info": None,
|
|
"id": "E",
|
|
"text": "Envoi atelier",
|
|
}
|
|
|
|
# invalid AS-TECH response
|
|
mocked_request.return_value = tests.utils.FakedResponse(content='{"foo":"bar"}', status_code=200)
|
|
response = app.get(endpoint + '?demand_id=000000000001234', status=200)
|
|
assert mocked_request.call_count == 2
|
|
assert response.json['err'] == 1
|
|
assert response.json['err_desc'].startswith('no position in response: ')
|
|
# invalid request
|
|
response = app.get(endpoint, status=400)
|
|
response = app.post_json(endpoint, status=405)
|
|
|
|
# get all possible positions
|
|
mocked_request.reset_mock()
|
|
endpoint = reverse(
|
|
"generic-endpoint",
|
|
kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "demand-all-positions"},
|
|
)
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=POSITIONS_RESPONSE, status_code=200)
|
|
response = app.get(endpoint, status=200)
|
|
assert mocked_request.call_count == 1
|
|
assert mocked_request.call_args[0][0] == 'get'
|
|
assert mocked_request.call_args[0][1].endswith('apicli/demande/positions')
|
|
assert response.json['err'] == 0
|
|
assert len(response.json['data']) == 11
|
|
assert response.json['data'][0] == {
|
|
"position": "A",
|
|
"positionLib": "En attente",
|
|
"color": "0, 0, 0",
|
|
"id": "A",
|
|
"text": "En attente",
|
|
}
|