adullact_pastell: add initial connector (#79105)
This commit is contained in:
parent
3cee8e4350
commit
c83228e375
|
|
@ -0,0 +1,77 @@
|
|||
# Generated by Django 3.2.18 on 2023-07-07 10:10
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
('base', '0030_resourcelog_base_resour_appname_298cbc_idx'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='AdullactPastell',
|
||||
fields=[
|
||||
(
|
||||
'id',
|
||||
models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
|
||||
),
|
||||
('title', models.CharField(max_length=50, verbose_name='Title')),
|
||||
('slug', models.SlugField(unique=True, verbose_name='Identifier')),
|
||||
('description', models.TextField(verbose_name='Description')),
|
||||
(
|
||||
'basic_auth_username',
|
||||
models.CharField(
|
||||
blank=True, max_length=128, verbose_name='Basic authentication username'
|
||||
),
|
||||
),
|
||||
(
|
||||
'basic_auth_password',
|
||||
models.CharField(
|
||||
blank=True, max_length=128, verbose_name='Basic authentication password'
|
||||
),
|
||||
),
|
||||
(
|
||||
'client_certificate',
|
||||
models.FileField(
|
||||
blank=True, null=True, upload_to='', verbose_name='TLS client certificate'
|
||||
),
|
||||
),
|
||||
(
|
||||
'trusted_certificate_authorities',
|
||||
models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'),
|
||||
),
|
||||
(
|
||||
'verify_cert',
|
||||
models.BooleanField(blank=True, default=True, verbose_name='TLS verify certificates'),
|
||||
),
|
||||
(
|
||||
'http_proxy',
|
||||
models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'),
|
||||
),
|
||||
(
|
||||
'api_base_url',
|
||||
models.URLField(
|
||||
help_text='Example: https://pastell.example.com/api/v2/',
|
||||
max_length=128,
|
||||
verbose_name='API base URL',
|
||||
),
|
||||
),
|
||||
('token', models.CharField(blank=True, max_length=128, verbose_name='API token')),
|
||||
(
|
||||
'users',
|
||||
models.ManyToManyField(
|
||||
blank=True,
|
||||
related_name='_adullact_pastell_adullactpastell_users_+',
|
||||
related_query_name='+',
|
||||
to='base.ApiUser',
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'Adullact Pastell',
|
||||
},
|
||||
),
|
||||
]
|
||||
|
|
@ -0,0 +1,234 @@
|
|||
# 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 base64
|
||||
from urllib import parse as urlparse
|
||||
|
||||
import requests
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.db import models
|
||||
from django.http import HttpResponse
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from passerelle.base.models import BaseResource, HTTPResource
|
||||
from passerelle.utils.api import endpoint
|
||||
from passerelle.utils.jsonresponse import APIError
|
||||
|
||||
FILE_OBJECT_PROPERTIES = {
|
||||
'title': _('File object'),
|
||||
'type': 'object',
|
||||
'properties': {
|
||||
'filename': {
|
||||
'type': 'string',
|
||||
'description': _('Filename'),
|
||||
},
|
||||
'content': {
|
||||
'type': 'string',
|
||||
'description': _('Content'),
|
||||
},
|
||||
'content_type': {
|
||||
'type': 'string',
|
||||
'description': _('Content type'),
|
||||
},
|
||||
},
|
||||
'required': ['filename', 'content'],
|
||||
}
|
||||
|
||||
|
||||
DOCUMENT_CREATION_SCHEMA = {
|
||||
'$schema': 'http://json-schema.org/draft-04/schema#',
|
||||
'type': 'object',
|
||||
'required': ['type'],
|
||||
'additionalProperties': True,
|
||||
'properties': {
|
||||
'type': {'type': 'string', 'description': _('Document type')},
|
||||
'file_field_name': {
|
||||
'type': 'string',
|
||||
'description': _('Document file\'s field name'),
|
||||
},
|
||||
'file': FILE_OBJECT_PROPERTIES,
|
||||
'filename': {
|
||||
'type': 'string',
|
||||
'description': _('Filename (takes precedence over filename in "file" object)'),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
DOCUMENT_FILE_UPLOAD_SCHEMA = {
|
||||
'$schema': 'http://json-schema.org/draft-04/schema#',
|
||||
'type': 'object',
|
||||
'required': ['file', 'file_field_name'],
|
||||
'additionalProperties': False,
|
||||
'properties': {
|
||||
'filename': {
|
||||
'type': 'string',
|
||||
'description': _('Filename (takes precedence over filename in "file" object)'),
|
||||
},
|
||||
'file': FILE_OBJECT_PROPERTIES,
|
||||
'file_field_name': {
|
||||
'type': 'string',
|
||||
'description': _('Document file\'s field name'),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class AdullactPastell(BaseResource, HTTPResource):
|
||||
api_base_url = models.URLField(
|
||||
max_length=128,
|
||||
verbose_name=_('API base URL'),
|
||||
help_text=_('Example: https://pastell.example.com/api/v2/'),
|
||||
)
|
||||
token = models.CharField(max_length=128, blank=True, verbose_name=_('API token'))
|
||||
|
||||
category = _('Business Process Connectors')
|
||||
|
||||
class Meta:
|
||||
verbose_name = _('Adullact Pastell')
|
||||
|
||||
def clean(self, *args, **kwargs):
|
||||
if not self.token and not self.basic_auth_username:
|
||||
raise ValidationError(_('API token or authentication username and password should be defined.'))
|
||||
return super().clean(*args, **kwargs)
|
||||
|
||||
def call(self, path, method='get', params=None, **kwargs):
|
||||
url = urlparse.urljoin(self.api_base_url, path)
|
||||
if self.token:
|
||||
kwargs.update({'headers': {'Authorization': 'Bearer: %s' % self.token}, 'auth': None})
|
||||
try:
|
||||
response = self.requests.request(url=url, method=method, params=params, **kwargs)
|
||||
response.raise_for_status()
|
||||
except (requests.Timeout, requests.RequestException) as e:
|
||||
raise APIError(str(e))
|
||||
return response
|
||||
|
||||
def check_status(self):
|
||||
try:
|
||||
response = self.call('version')
|
||||
except APIError as e:
|
||||
raise Exception('Pastell server is down: %s' % e)
|
||||
return {'data': response.json()}
|
||||
|
||||
def upload_file(self, entity_id, document_id, file_field_name, data, **kwargs):
|
||||
filename = kwargs.get('filename') or data['filename']
|
||||
file_data = {
|
||||
'file_content': (
|
||||
filename,
|
||||
base64.b64decode(data['content']),
|
||||
data.get('content_type'),
|
||||
)
|
||||
}
|
||||
|
||||
return self.call(
|
||||
'entite/%s/document/%s/file/%s' % (entity_id, document_id, file_field_name),
|
||||
'post',
|
||||
files=file_data,
|
||||
data={'file_name': filename},
|
||||
)
|
||||
|
||||
@endpoint(
|
||||
description=_('List entities'),
|
||||
datasource=True,
|
||||
)
|
||||
def entities(self, request):
|
||||
data = []
|
||||
response = self.call('entite')
|
||||
for item in response.json():
|
||||
item['id'] = item['id_e']
|
||||
item['text'] = item['denomination']
|
||||
data.append(item)
|
||||
return {'data': data}
|
||||
|
||||
@endpoint(
|
||||
description=_('List entity documents'),
|
||||
parameters={'entity_id': {'description': _('Entity ID'), 'example_value': '42'}},
|
||||
datasource=True,
|
||||
)
|
||||
def documents(self, request, entity_id):
|
||||
if request.GET.get('id'):
|
||||
response = self.call('entite/%s/document/%s' % (entity_id, request.GET['id']))
|
||||
return {'data': response.json()}
|
||||
|
||||
data = []
|
||||
response = self.call('entite/%s/document' % entity_id)
|
||||
for item in response.json():
|
||||
item['id'] = item['id_d']
|
||||
item['text'] = item['titre']
|
||||
data.append(item)
|
||||
return {'data': data}
|
||||
|
||||
@endpoint(
|
||||
post={
|
||||
'description': _('Create a document for an entity'),
|
||||
'request_body': {'schema': {'application/json': DOCUMENT_CREATION_SCHEMA}},
|
||||
},
|
||||
name="create-document",
|
||||
parameters={
|
||||
'entity_id': {'description': _('Entity ID'), 'example_value': '42'},
|
||||
},
|
||||
)
|
||||
def create_document(self, request, entity_id, post_data):
|
||||
file_data = post_data.pop('file', None)
|
||||
file_field_name = post_data.pop('file_field_name', None)
|
||||
|
||||
# create document
|
||||
response = self.call('entite/%s/document' % entity_id, 'post', params=post_data)
|
||||
document_id = response.json()['id_d']
|
||||
|
||||
# update it with other attributes
|
||||
response = self.call('entite/%s/document/%s' % (entity_id, document_id), 'patch', params=post_data)
|
||||
|
||||
# upload file if it's filled
|
||||
if file_field_name and file_data:
|
||||
self.upload_file(entity_id, document_id, file_field_name, file_data, **post_data)
|
||||
|
||||
return {'data': response.json()}
|
||||
|
||||
@endpoint(
|
||||
post={
|
||||
'description': _('Upload a file to a document'),
|
||||
'request_body': {'schema': {'application/json': DOCUMENT_FILE_UPLOAD_SCHEMA}},
|
||||
},
|
||||
name="upload-document-file",
|
||||
parameters={
|
||||
'entity_id': {'description': _('Entity ID'), 'example_value': '42'},
|
||||
'document_id': {'description': _('Document ID'), 'example_value': 'hDWtdSC'},
|
||||
},
|
||||
)
|
||||
def upload_document_file(self, request, entity_id, document_id, post_data):
|
||||
file_field_name = post_data.pop('file_field_name')
|
||||
file_data = post_data.pop('file')
|
||||
response = self.upload_file(entity_id, document_id, file_field_name, file_data, **post_data)
|
||||
return {'data': response.json()}
|
||||
|
||||
@endpoint(
|
||||
description=_('Get document\'s file'),
|
||||
name="get-document-file",
|
||||
parameters={
|
||||
'entity_id': {'description': _('Entity ID'), 'example_value': '42'},
|
||||
'document_id': {'description': _('Document ID'), 'example_value': 'hDWtdSC'},
|
||||
'field_name': {
|
||||
'description': _('Document file\'s field name'),
|
||||
'example_value': 'document',
|
||||
},
|
||||
},
|
||||
)
|
||||
def get_document_file(self, request, entity_id, document_id, field_name):
|
||||
document = self.call('entite/%s/document/%s/file/%s' % (entity_id, document_id, field_name))
|
||||
response = HttpResponse(document.content, content_type=document.headers['Content-Type'])
|
||||
response['Content-Disposition'] = document.headers['Content-disposition']
|
||||
return response
|
||||
|
|
@ -126,6 +126,7 @@ INSTALLED_APPS = (
|
|||
'passerelle.sms',
|
||||
# connectors
|
||||
'passerelle.apps.actesweb',
|
||||
'passerelle.apps.adullact_pastell',
|
||||
'passerelle.apps.airquality',
|
||||
'passerelle.apps.api_entreprise',
|
||||
'passerelle.apps.api_particulier',
|
||||
|
|
|
|||
|
|
@ -0,0 +1,345 @@
|
|||
# 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'
|
||||
Loading…
Reference in New Issue