368 lines
11 KiB
Python
368 lines
11 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# Copyright (C) 2023 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 pytest
|
|
import responses
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.urls import reverse
|
|
|
|
from passerelle.apps.adullact_pastell.models import AdullactPastell
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
|
|
|
|
@pytest.fixture()
|
|
def setup(db):
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
conn = AdullactPastell.objects.create(
|
|
api_base_url='http://example.com/api/v2/',
|
|
basic_auth_username='admin',
|
|
basic_auth_password='admin',
|
|
slug='test',
|
|
)
|
|
obj_type = ContentType.objects.get_for_model(conn)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=conn.pk
|
|
)
|
|
return conn
|
|
|
|
|
|
@responses.activate
|
|
def test_auth_headers(app, setup):
|
|
responses.add(
|
|
responses.GET,
|
|
'http://example.com/api/v2/version',
|
|
json={'version': '3.0.4', 'revision': 54322},
|
|
status=200,
|
|
)
|
|
setup.check_status()
|
|
assert len(responses.calls) == 1
|
|
assert 'Basic ' in responses.calls[0].request.headers['Authorization']
|
|
|
|
setup.token = '12345'
|
|
setup.save()
|
|
setup.check_status()
|
|
assert len(responses.calls) == 2
|
|
assert responses.calls[1].request.headers['Authorization'] == 'Bearer: 12345'
|
|
|
|
|
|
@responses.activate
|
|
def test_service_down_status(app, setup):
|
|
responses.add(
|
|
responses.GET,
|
|
'http://example.com/api/v2/version',
|
|
status=404,
|
|
)
|
|
with pytest.raises(Exception):
|
|
setup.check_status()
|
|
|
|
|
|
@responses.activate
|
|
def test_list_entities(app, setup):
|
|
responses.add(
|
|
responses.GET,
|
|
'http://example.com/api/v2/entite',
|
|
json=[
|
|
{
|
|
'id_e': '7',
|
|
'denomination': 'Publik',
|
|
'siren': '198307662',
|
|
'type': 'collectivite',
|
|
'centre_de_gestion': '0',
|
|
'entite_mere': '1',
|
|
}
|
|
],
|
|
status=200,
|
|
)
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'adullact-pastell', 'endpoint': 'entities', 'slug': setup.slug},
|
|
)
|
|
resp = app.get(url)
|
|
data = resp.json['data']
|
|
assert data[0]['id'] == '7'
|
|
assert data[0]['text'] == 'Publik'
|
|
resp = app.get(url, params={'id': 42})
|
|
assert resp.json['data'] == []
|
|
|
|
|
|
@responses.activate
|
|
def test_list_documents(app, setup):
|
|
responses.add(
|
|
responses.GET,
|
|
'http://example.com/api/v2/entite/7/document',
|
|
json=[
|
|
{
|
|
'id_d': 'MNYDNCa',
|
|
'id_e': '7',
|
|
'type': 'publik',
|
|
'titre': 'TestConnecteur',
|
|
'creation': '2023-06-27 18:02:04',
|
|
'modification': '2023-06-27 18:02:28',
|
|
'denomination': 'Publik',
|
|
'entite_base': 'Publik',
|
|
}
|
|
],
|
|
status=200,
|
|
)
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'adullact-pastell', 'endpoint': 'documents', 'slug': setup.slug},
|
|
)
|
|
resp = app.get(url, params={'entity_id': '7'})
|
|
data = resp.json['data']
|
|
assert data[0]['id'] == 'MNYDNCa'
|
|
assert data[0]['text'] == 'TestConnecteur'
|
|
|
|
|
|
@responses.activate
|
|
def test_get_document_details(app, setup):
|
|
responses.add(
|
|
responses.GET,
|
|
'http://example.com/api/v2/entite/7/document/MNYDNCa',
|
|
json={
|
|
'info': {
|
|
'id_d': 'MNYDNCa',
|
|
'type': 'publik',
|
|
'titre': 'TestConnecteur',
|
|
'creation': '2023-06-27 18:02:04',
|
|
'modification': '2023-06-27 18:02:28',
|
|
},
|
|
'data': {
|
|
'envoi_signature': 'checked',
|
|
'envoi_iparapheur': '1',
|
|
'nom_dossier': 'TestConnecteur',
|
|
'iparapheur_type': 'Type Publik',
|
|
'iparapheur_sous_type': 'SPublik',
|
|
},
|
|
},
|
|
status=200,
|
|
)
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'adullact-pastell', 'endpoint': 'documents', 'slug': setup.slug},
|
|
)
|
|
resp = app.get(url, params={'entity_id': '7', 'id': 'MNYDNCa'})
|
|
data = resp.json['data']
|
|
assert data['data']['nom_dossier'] == 'TestConnecteur'
|
|
assert data['info']['id_d'] == 'MNYDNCa'
|
|
|
|
|
|
@responses.activate
|
|
def test_create_document(app, setup):
|
|
responses.add(
|
|
responses.POST,
|
|
'http://example.com/api/v2/entite/7/document',
|
|
json={
|
|
'info': {
|
|
'id_d': '67WaYzM',
|
|
'type': 'publik',
|
|
'titre': '',
|
|
'creation': '2023-06-27 17:25:54',
|
|
'modification': '2023-06-27 17:25:54',
|
|
},
|
|
'data': {'envoi_signature': 'checked', 'envoi_iparapheur': '1'},
|
|
'action_possible': ['modification', 'supression'],
|
|
'action-possible': ['modification', 'supression'],
|
|
'last_action': {
|
|
'action': 'creation',
|
|
'message': 'Cr\u00e9ation du document',
|
|
'date': '2023-06-27 17:25:54',
|
|
},
|
|
'id_d': '67WaYzM',
|
|
},
|
|
status=200,
|
|
)
|
|
responses.add(
|
|
responses.PATCH,
|
|
'http://example.com/api/v2/entite/7/document/67WaYzM',
|
|
json={
|
|
'content': {
|
|
'info': {
|
|
'id_d': 'MNYDNCa',
|
|
'type': 'publik',
|
|
'titre': 'TestConnecteur',
|
|
'creation': '2023-06-27 18:02:04',
|
|
'modification': '2023-06-27 18:02:28',
|
|
},
|
|
'data': {
|
|
'envoi_signature': 'checked',
|
|
'envoi_iparapheur': '1',
|
|
'nom_dossier': 'TestConnecteu',
|
|
'iparapheur_type': 'Type Publik',
|
|
'iparapheur_sous_type': 'SPublik',
|
|
},
|
|
},
|
|
'result': 'ok',
|
|
'formulaire_ok': 0,
|
|
'message': 'Le formulaire est incomplet : le champ \u00abDocument\u00bb est obligatoire.',
|
|
},
|
|
status=200,
|
|
)
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'adullact-pastell', 'endpoint': 'create-document', 'slug': setup.slug},
|
|
)
|
|
resp = app.post_json(
|
|
'%s?entity_id=7' % url,
|
|
params={
|
|
'type': 'publik',
|
|
'nom_dossier': 'TestConnecteur',
|
|
'iparapheur_type': 'Type Publik',
|
|
'iparapheur_sous_type': 'SPublik',
|
|
},
|
|
)
|
|
assert len(responses.calls) == 2
|
|
assert resp.json['data']['result'] == 'ok'
|
|
|
|
responses.add(
|
|
responses.POST,
|
|
'http://example.com/api/v2/entite/7/document/67WaYzM/file/document',
|
|
json={
|
|
'result': 'ok',
|
|
'formulaire_ok': 1,
|
|
'message': '',
|
|
'err': 0,
|
|
},
|
|
status=201,
|
|
)
|
|
resp = app.post_json(
|
|
'%s?entity_id=7' % url,
|
|
params={
|
|
'type': 'publik',
|
|
'nom_dossier': 'TestConnecteur',
|
|
'iparapheur_type': 'Type Publik',
|
|
'iparapheur_sous_type': 'SPublik',
|
|
'file_field_name': 'document',
|
|
'file': {
|
|
'filename': 'test.txt',
|
|
'content': 'dGVzdA==',
|
|
},
|
|
},
|
|
)
|
|
assert len(responses.calls) == 5
|
|
assert resp.json['data']['result'] == 'ok'
|
|
|
|
|
|
@responses.activate
|
|
def test_add_document_file(app, setup):
|
|
responses.add(
|
|
responses.POST,
|
|
'http://example.com/api/v2/entite/7/document/13fgg3E/file/document',
|
|
json={
|
|
'content': {
|
|
'info': {
|
|
'id_d': '13fgg3E',
|
|
'type': 'publik',
|
|
'titre': 'DocumentViaAPI',
|
|
'creation': '2023-07-03 17:35:16',
|
|
'modification': '2023-07-03 17:35:16',
|
|
},
|
|
'data': {
|
|
'envoi_signature': 'checked',
|
|
'envoi_iparapheur': '1',
|
|
'nom_dossier': 'DocumentViaAPI',
|
|
'iparapheur_type': 'Type Publik',
|
|
'iparapheur_sous_type': 'SPublik',
|
|
'document': ['TestSerghei.pdf'],
|
|
},
|
|
},
|
|
'result': 'ok',
|
|
'formulaire_ok': 1,
|
|
'message': '',
|
|
'err': 0,
|
|
},
|
|
status=201,
|
|
)
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'adullact-pastell', 'endpoint': 'upload-document-file', 'slug': setup.slug},
|
|
)
|
|
resp = app.post_json(
|
|
'%s?entity_id=7&document_id=13fgg3E' % url,
|
|
params={
|
|
'file_field_name': 'document',
|
|
'file': {
|
|
'filename': 'test.txt',
|
|
'content': 'dGVzdA==',
|
|
'content_type': 'text/plain',
|
|
},
|
|
},
|
|
)
|
|
assert b'filename="test.txt"' in responses.calls[0].request.body
|
|
assert resp.json['data']['result'] == 'ok'
|
|
resp = app.post_json(
|
|
'%s?entity_id=7&document_id=13fgg3E' % url,
|
|
params={
|
|
'file_field_name': 'document',
|
|
'filename': 'new.txt',
|
|
'file': {
|
|
'filename': 'test.txt',
|
|
'content': 'dGVzdA==',
|
|
'content_type': 'text/plain',
|
|
},
|
|
},
|
|
)
|
|
assert b'filename="new.txt"' in responses.calls[1].request.body
|
|
|
|
|
|
@responses.activate
|
|
def test_get_document_file(app, setup):
|
|
responses.add(
|
|
responses.GET,
|
|
'http://example.com/api/v2/entite/7/document/13fgg3E/file/document',
|
|
body='test',
|
|
content_type='text/plain',
|
|
headers={'Content-disposition': 'attachment; filename=test.txt'},
|
|
status=200,
|
|
)
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'adullact-pastell', 'endpoint': 'get-document-file', 'slug': setup.slug},
|
|
)
|
|
resp = app.get(
|
|
'%s?entity_id=7&document_id=13fgg3E&field_name=document' % url,
|
|
)
|
|
assert resp.headers.get('Content-Disposition') == 'attachment; filename=test.txt'
|
|
assert resp.headers.get('Content-Type') == 'text/plain'
|
|
assert resp.body == b'test'
|
|
|
|
|
|
@responses.activate
|
|
def test_run_document_action(app, setup):
|
|
responses.add(
|
|
responses.POST,
|
|
'http://example.com/api/v2/entite/7/document/13fgg3E/action/orientation',
|
|
body='{"result": true, "message": "sélection automatique de l\'action suivante"}',
|
|
status=200,
|
|
)
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'adullact-pastell', 'endpoint': 'run-document-action', 'slug': setup.slug},
|
|
)
|
|
resp = app.post_json(
|
|
'%s?entity_id=7&document_id=13fgg3E' % url,
|
|
params={'action_name': 'orientation'},
|
|
)
|
|
print(resp.request.url)
|
|
assert resp.json['data']['result']
|
|
assert resp.json['data']['message']
|