524 lines
22 KiB
Python
524 lines
22 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"}]
|
|
"""
|
|
|
|
VIEWS_RESPONSE = """
|
|
{"views":[{"apivId":"7","apivCode":"ASTECH_FORMDYN","apivNom":"Formulaires dynamiques - champs de saisie"},
|
|
{"apivId":"1","apivCode":"ASTECH_BIENS","apivNom":"Liste des biens"}]}
|
|
"""
|
|
|
|
COLUMNS_RESPONSE = """
|
|
{"columns":[{"code":"BIEN_ID","des":"Identifiant du bien AS-TECH","type":"NUM","length":"18"},
|
|
{"code":"ANCETREID","des":"Identifiant de l\u0027anc\u00eatre du bien","type":"NUM","length":""}]}
|
|
"""
|
|
|
|
RESULTS_RESPONSE = """[{"BIEN_ID": "2219", "CODE_BIEN": "AC-849-YE", "DESIGNATION": "RENAULT KANGOO"},
|
|
{"BIEN_ID": "2220", "CODE_BIEN": "AC-933-EA", "DESIGNATION": "RENAULT MASTER"},
|
|
{"BIEN_ID": "2221", "CODE_BIEN": "AC-955-SE", "DESIGNATION": "RENAULT KANGOO"}]
|
|
"""
|
|
|
|
|
|
@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'},
|
|
]
|
|
|
|
mocked_request.return_value = tests.utils.FakedResponse(content='[]', status_code=200)
|
|
response = app.get(endpoint)
|
|
assert response.json['err'] == 1
|
|
assert response.json['err_desc'] == 'Invalid response: []'
|
|
|
|
|
|
@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
|
|
|
|
mocked_request.return_value = tests.utils.FakedResponse(content='[]', status_code=200)
|
|
response = app.get(endpoint)
|
|
assert response.json['err'] == 1
|
|
assert response.json['err_desc'] == 'Invalid response: []'
|
|
|
|
|
|
@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',
|
|
}
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.request')
|
|
@mock.patch('passerelle.apps.astech.models.ASTech.get_authorization')
|
|
def test_list_views(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': 'list-views'},
|
|
)
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=VIEWS_RESPONSE, status_code=200)
|
|
response = app.get(endpoint)
|
|
assert mocked_request.call_args[0][0] == 'get'
|
|
assert mocked_request.call_args[0][1].endswith('apicli/data/views')
|
|
assert response.json['data']
|
|
for r in response.json['data']:
|
|
assert 'id' in r
|
|
assert 'text' in r
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.request')
|
|
@mock.patch('passerelle.apps.astech.models.ASTech.get_authorization')
|
|
def test_view_columns(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': 'get-view-columns'},
|
|
)
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=COLUMNS_RESPONSE, status_code=200)
|
|
response = app.get(endpoint, params={'code': 'ASTECH_BIENS'})
|
|
assert mocked_request.call_args[0][0] == 'get'
|
|
assert mocked_request.call_args[0][1].endswith('apicli/data/ASTECH_BIENS/columns')
|
|
assert response.json['data']
|
|
for r in response.json['data']:
|
|
assert 'id' in r
|
|
assert 'text' in r
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.request')
|
|
@mock.patch('passerelle.apps.astech.models.ASTech.get_authorization')
|
|
def test_view_data(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': 'get-view-data'},
|
|
)
|
|
mocked_request.return_value = tests.utils.FakedResponse(content=RESULTS_RESPONSE, status_code=200)
|
|
response = app.get(
|
|
endpoint, params={'code': 'ASTECH_BIENS', 'id_column': 'BIEN_ID', 'text_column': 'DESIGNATION'}
|
|
)
|
|
assert mocked_request.call_args[0][0] == 'post'
|
|
assert mocked_request.call_args[0][1].endswith('apicli/data/ASTECH_BIENS/results')
|
|
assert response.json['data']
|
|
for r in response.json['data']:
|
|
assert 'id' in r
|
|
assert 'text' in r
|
|
|
|
response = app.get(
|
|
endpoint,
|
|
params={'code': 'ASTECH_BIENS', 'id_column': 'BIEN_ID', 'text_column': 'DESIGNATION', 'id': 2221},
|
|
)
|
|
assert len(response.json['data']) == 1
|
|
|
|
response = app.get(
|
|
endpoint,
|
|
params={'code': 'ASTECH_BIENS', 'id_column': 'BIEN_ID', 'text_column': 'DESIGNATION', 'q': 'KANGOO'},
|
|
)
|
|
assert len(response.json['data']) == 2
|