wip/73841-greco-rest (#73841) #69
|
@ -101,6 +101,7 @@ class FormData:
|
|||
|
||||
# extract/create/validate fields according to schema
|
||||
self.fields = {}
|
||||
self.json = {}
|
||||
for fieldname in schema:
|
||||
if isinstance(fieldname, tuple):
|
||||
value = values.get(fieldname[0])
|
||||
|
@ -114,6 +115,22 @@ class FormData:
|
|||
value = values.get(fieldname)
|
||||
if value is not None:
|
||||
self.fields[fieldname] = value
|
||||
if isinstance(value, datetime):
|
||||
value = value.strftime('%Y-%m-%dT%H:%M:%S')
|
||||
if '_' not in fieldname:
|
||||
self.json[fieldname] = value
|
||||
continue
|
||||
key, subkey = fieldname.split('_')
|
||||
if key not in self.json:
|
||||
self.json[key] = {}
|
||||
self.json[key][subkey] = value
|
||||
|
||||
# https://dev.entrouvert.org/issues/75259
|
||||
danger = self.json.get('danger', '')
|
||||
if danger:
|
||||
self.json['danger'] = 'true'
|
||||
else:
|
||||
self.json['danger'] = 'false'
|
||||
|
||||
# extract attachments
|
||||
self.attachments = []
|
||||
|
|
|
@ -37,7 +37,7 @@ class Migration(migrations.Migration):
|
|||
('application', models.CharField(max_length=200, verbose_name='Application identifier')),
|
||||
('token_url', models.URLField(max_length=256, verbose_name='Token URL')),
|
||||
('token_authorization', models.CharField(max_length=128, verbose_name='Token Authorization')),
|
||||
('wsdl_url', models.CharField(max_length=256, verbose_name='WSDL URL')),
|
||||
('wsdl_url', models.CharField(max_length=256, verbose_name='WSDL or REST URL')),
|
||||
(
|
||||
'verify_cert',
|
||||
models.BooleanField(default=True, verbose_name='Check HTTPS Certificate validity'),
|
||||
|
|
|
@ -0,0 +1,23 @@
|
|||
# Generated by Django 2.2.26 on 2023-02-01 11:11
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('greco', '0003_remove_greco_log_level'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AddField(
|
||||
model_name='greco',
|
||||
name='api_type',
|
||||
field=models.CharField(
|
||||
choices=[('soap', 'SOAP'), ('rest', 'REST')],
|
||||
default='soap',
|
||||
max_length=4,
|
||||
verbose_name='API Type',
|
||||
),
|
||||
),
|
||||
]
|
|
@ -14,12 +14,15 @@
|
|||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
import urllib
|
||||
from email.mime.base import MIMEBase
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
|
||||
import requests
|
||||
import suds.sudsobject
|
||||
from django.core.cache import cache
|
||||
from django.db import models
|
||||
|
@ -34,6 +37,12 @@ from passerelle.utils.api import APIError, endpoint
|
|||
|
||||
from .formdata import CREATION_SCHEMA, FormData, list_schema_fields
|
||||
|
||||
API_TYPES = (
|
||||
('soap', 'SOAP'),
|
||||
('rest', 'REST'),
|
||||
)
|
||||
|
||||
|
||||
# taken from https://lsimons.wordpress.com/2011/03/17/stripping-illegal-characters-out-of-xml-in-python/
|
||||
_illegal_xml_chars_RE = re.compile('[\x00-\x08\x0b\x0c\x0e-\x1F\uD800-\uDFFF\uFFFE\uFFFF]')
|
||||
|
||||
|
@ -69,8 +78,9 @@ class Greco(BaseResource):
|
|||
application = models.CharField(_('Application identifier'), max_length=200)
|
||||
token_url = models.URLField(_('Token URL'), max_length=256)
|
||||
token_authorization = models.CharField(_('Token Authorization'), max_length=128)
|
||||
wsdl_url = models.CharField(_('WSDL URL'), max_length=256) # not URLField, it can be file://
|
||||
wsdl_url = models.CharField(_('WSDL or REST URL'), max_length=256) # not URLField, it can be file://
|
||||
verify_cert = models.BooleanField(default=True, verbose_name=_('Check HTTPS Certificate validity'))
|
||||
api_type = models.CharField(_('API Type'), max_length=4, choices=API_TYPES, default='soap')
|
||||
|
||||
hide_description_fields = ['token_authorization']
|
||||
category = _('Business Process Connectors')
|
||||
|
@ -78,6 +88,10 @@ class Greco(BaseResource):
|
|||
class Meta:
|
||||
verbose_name = _('GRECO Webservices')
|
||||
|
||||
@property
|
||||
def use_soap(self):
|
||||
return self.api_type == 'soap'
|
||||
|
||||
def get_token(self, renew=False):
|
||||
cache_key = 'greco-%s-token' % self.id
|
||||
if not renew:
|
||||
|
@ -215,8 +229,60 @@ class Greco(BaseResource):
|
|||
|
||||
return Client(url=self.wsdl_url, transport=Transport(self, attachments))
|
||||
|
||||
def _rest_call(self, path, method='get', json_data=None, params=None, files=None):
|
||||
headers = {'Authorization': self.get_token()}
|
||||
url = urllib.parse.urljoin(self.wsdl_url, path)
|
||||
try:
|
||||
resp = self.requests.request(
|
||||
method=method,
|
||||
url=url,
|
||||
headers=headers,
|
||||
json=json_data,
|
||||
params=params,
|
||||
files=files,
|
||||
)
|
||||
except (requests.Timeout, requests.RequestException) as e:
|
||||
raise APIError(str(e))
|
||||
if resp.status_code == 401:
|
||||
# ask for a new token, and retry
|
||||
headers['Authorization'] = self.get_token(renew=True)
|
||||
try:
|
||||
resp = self.requests.request(
|
||||
method=method,
|
||||
url=url,
|
||||
headers=headers,
|
||||
json=json_data,
|
||||
params=params,
|
||||
files=files,
|
||||
)
|
||||
except (requests.Timeout, requests.RequestException) as e:
|
||||
raise APIError(str(e))
|
||||
|
||||
try:
|
||||
resp.raise_for_status()
|
||||
except requests.RequestException as e:
|
||||
try:
|
||||
err_data = resp.json()
|
||||
except (json.JSONDecodeError, requests.RequestException):
|
||||
err_data = {'response_text': resp.text}
|
||||
raise APIError(str(e), data=err_data)
|
||||
|
||||
content_type = resp.headers.get('Content-Type')
|
||||
if content_type and content_type.startswith('application/json'):
|
||||
try:
|
||||
return resp.json()
|
||||
except (json.JSONDecodeError, requests.RequestException) as e:
|
||||
err_data = {'response_text': resp.text}
|
||||
raise APIError(str(e), data=err_data)
|
||||
|
||||
raise APIError(resp.text)
|
||||
|
||||
def check_status(self):
|
||||
if self.get_client().service.communicationTest('ping') is None:
|
||||
if self.use_soap:
|
||||
response = self.get_client().service.communicationTest('ping')
|
||||
else:
|
||||
response = self._rest_call('communicationTest/ping').get('reponse')
|
||||
if not response:
|
||||
raise Exception('empty answer to communication test')
|
||||
|
||||
@endpoint(
|
||||
|
@ -227,8 +293,12 @@ class Greco(BaseResource):
|
|||
display_order=1,
|
||||
)
|
||||
def ping(self, request):
|
||||
resp = self.get_client().service.communicationTest('ping')
|
||||
if resp is None:
|
||||
if self.use_soap:
|
||||
resp = self.get_client().service.communicationTest('ping')
|
||||
else:
|
||||
json_data = self._rest_call('communicationTest/ping')
|
||||
resp = json_data.get('reponse')
|
||||
if not resp:
|
||||
raise APIError('empty response from communicationTest()')
|
||||
return {'data': resp}
|
||||
|
||||
|
@ -261,16 +331,32 @@ Response :
|
|||
formdata = FormData(json.loads(request.body), CREATION_SCHEMA)
|
||||
except ValueError as e:
|
||||
raise ParameterTypeError(str(e))
|
||||
# create suds object from formdata
|
||||
client = self.get_client(formdata.attachments)
|
||||
creation = client.factory.create('DemandeCreation')
|
||||
creation.application = self.application
|
||||
fill_sudsobject_with_dict(creation, formdata.fields)
|
||||
# send it to "creer"
|
||||
resp = client.service.creer(creation)
|
||||
if resp is None:
|
||||
raise APIError('empty response from creer()')
|
||||
return {'data': sudsobject_to_dict(resp)}
|
||||
if self.use_soap:
|
||||
# create suds object from formdata
|
||||
client = self.get_client(formdata.attachments)
|
||||
creation = client.factory.create('DemandeCreation')
|
||||
creation.application = self.application
|
||||
fill_sudsobject_with_dict(creation, formdata.fields)
|
||||
# send it to "creer"
|
||||
resp = client.service.creer(creation)
|
||||
if resp is None:
|
||||
raise APIError('empty response from creer()')
|
||||
data = sudsobject_to_dict(resp)
|
||||
else:
|
||||
files = []
|
||||
for num, attachment in enumerate(formdata.attachments):
|
||||
filename = attachment.get('filename') or 'file%s.bin' % num
|
||||
content = base64.b64decode(attachment.get('content') or '')
|
||||
content_type = attachment.get('content_type') or 'application/octet-stream'
|
||||
files.append(('files', (filename, content, content_type)))
|
||||
files.append(
|
||||
(
|
||||
'creerRequest',
|
||||
(None, io.BytesIO(json.dumps(formdata.json).encode('utf-8')), 'application/json'),
|
||||
)
|
||||
)
|
||||
data = self._rest_call('creer', method='post', files=files)
|
||||
return {'data': data}
|
||||
|
||||
@classmethod
|
||||
def creation_fields(cls):
|
||||
|
@ -289,15 +375,18 @@ Response :
|
|||
display_order=3,
|
||||
)
|
||||
def status(self, request, idgreco, iddemande=None):
|
||||
resp = self.get_client().service.consulter(
|
||||
{
|
||||
'idgreco': idgreco,
|
||||
'iddemande': iddemande,
|
||||
}
|
||||
)
|
||||
if resp is None:
|
||||
raise APIError('empty response from status()')
|
||||
return {'data': sudsobject_to_dict(resp)}
|
||||
params = {
|
||||
'idgreco': idgreco,
|
||||
'iddemande': iddemande,
|
||||
}
|
||||
if self.use_soap:
|
||||
resp = self.get_client().service.consulter(params)
|
||||
if resp is None:
|
||||
raise APIError('empty response from status()')
|
||||
data = sudsobject_to_dict(resp)
|
||||
else:
|
||||
data = self._rest_call('consulter', params=params)
|
||||
return {'data': data}
|
||||
|
||||
@endpoint(
|
||||
perm='can_access',
|
||||
|
@ -313,10 +402,14 @@ Response :
|
|||
}
|
||||
if code:
|
||||
params['taCode'] = code
|
||||
resp = self.get_client().service.getMail(params)
|
||||
if resp is None:
|
||||
raise APIError('empty response from consulter()')
|
||||
return {'data': sudsobject_to_dict(resp)}
|
||||
if self.use_soap:
|
||||
resp = self.get_client().service.getMail(params)
|
||||
if resp is None:
|
||||
raise APIError('empty response from consulter()')
|
||||
data = sudsobject_to_dict(resp)
|
||||
else:
|
||||
data = self._rest_call('getMail', params=params)
|
||||
return {'data': data}
|
||||
|
||||
@endpoint(
|
||||
perm='can_access',
|
||||
|
@ -335,16 +428,19 @@ Response :
|
|||
idgreco = payload.get('idgreco') or idgreco
|
||||
iddemande = payload.get('iddemande') or iddemande
|
||||
information = payload.get('information') or information
|
||||
resp = self.get_client().service.ajouterComplementInformation(
|
||||
{
|
||||
'idgreco': idgreco,
|
||||
'iddemande': iddemande,
|
||||
'complementInfo': information,
|
||||
}
|
||||
)
|
||||
if resp is None:
|
||||
raise APIError('empty response from ajouterComplementInformation()')
|
||||
return {'data': sudsobject_to_dict(resp)}
|
||||
json_body = {
|
||||
'idgreco': idgreco,
|
||||
'iddemande': iddemande,
|
||||
'complementInfo': information,
|
||||
}
|
||||
if self.use_soap:
|
||||
resp = self.get_client().service.ajouterComplementInformation(json_body)
|
||||
if resp is None:
|
||||
raise APIError('empty response from ajouterComplementInformation()')
|
||||
data = sudsobject_to_dict(resp)
|
||||
else:
|
||||
data = self._rest_call('ajouterComplementInformation', method='post', json_data=json_body)
|
||||
return {'data': data}
|
||||
|
||||
@endpoint(
|
||||
perm='can_access',
|
||||
|
@ -362,16 +458,19 @@ Response :
|
|||
idgreco = payload.get('idgreco') or idgreco
|
||||
iddemande = payload.get('iddemande') or iddemande
|
||||
comment = payload.get('comment') or comment
|
||||
resp = self.get_client().service.relancer(
|
||||
{
|
||||
'idgreco': idgreco,
|
||||
'iddemande': iddemande,
|
||||
'commentaire': comment,
|
||||
}
|
||||
)
|
||||
if resp is None:
|
||||
raise APIError('empty response from relancer()')
|
||||
return {'data': sudsobject_to_dict(resp)}
|
||||
params = {
|
||||
'idgreco': idgreco,
|
||||
'iddemande': iddemande,
|
||||
'commentaire': comment,
|
||||
}
|
||||
if self.use_soap:
|
||||
resp = self.get_client().service.relancer(params)
|
||||
if resp is None:
|
||||
raise APIError('empty response from relancer()')
|
||||
data = sudsobject_to_dict(resp)
|
||||
else:
|
||||
data = self._rest_call('relancer', params=params)
|
||||
return {'data': data}
|
||||
|
||||
@endpoint(
|
||||
perm='can_access',
|
||||
|
@ -389,13 +488,16 @@ Response :
|
|||
idgreco = payload.get('idgreco')
|
||||
iddemande = payload.get('iddemande')
|
||||
nbr = payload.get('nbconfirmation')
|
||||
resp = self.get_client().service.confirmer(
|
||||
{
|
||||
'idGreco': idgreco,
|
||||
'idDemande': iddemande,
|
||||
'nbconfirmation': nbr,
|
||||
}
|
||||
)
|
||||
if resp is None:
|
||||
raise APIError('empty response from confirmer()')
|
||||
return {'data': sudsobject_to_dict(resp)}
|
||||
params = {
|
||||
'idGreco': idgreco,
|
||||
'idDemande': iddemande,
|
||||
'nbconfirmation': nbr,
|
||||
}
|
||||
if self.use_soap:
|
||||
resp = self.get_client().service.confirmer(params)
|
||||
if resp is None:
|
||||
raise APIError('empty response from confirmer()')
|
||||
data = sudsobject_to_dict(resp)
|
||||
else:
|
||||
data = self._rest_call('confirmer', params=params)
|
||||
return {'data': data}
|
||||
|
|
|
@ -14,15 +14,18 @@
|
|||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import xml.etree.ElementTree as ET
|
||||
from base64 import b64encode
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
import responses
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.urls import reverse
|
||||
from django.utils.encoding import force_str
|
||||
from requests_toolbelt import MultipartDecoder
|
||||
|
||||
import tests.utils
|
||||
from passerelle.base.models import AccessRight, ApiUser
|
||||
|
@ -104,6 +107,7 @@ CREATE_PAYLOAD = {
|
|||
'transmetteur_service': 'GNM',
|
||||
'transmetteur_nom': 'Vertommen',
|
||||
'transmetteur_prenom': 'Agent 15',
|
||||
'danger': False,
|
||||
}
|
||||
|
||||
|
||||
|
@ -126,6 +130,14 @@ def conn():
|
|||
return connector
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rest_conn(conn):
|
||||
conn.wsdl_url = 'http://greco.example.net'
|
||||
conn.api_type = 'rest'
|
||||
conn.save()
|
||||
return conn
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.post', side_effect=(RESP_503,))
|
||||
def test_greco_no_api_key(mocked_post, app, conn):
|
||||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'ping', 'slug': conn.slug})
|
||||
|
@ -204,6 +216,51 @@ def test_greco_create_ok(mocked_post, app, conn):
|
|||
}
|
||||
|
||||
|
||||
def test_greco_rest_create_ok(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'create', 'slug': rest_conn.slug}
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.post(
|
||||
'http://greco.example.net/creer',
|
||||
status=200,
|
||||
json={
|
||||
'iddemande': 'MDFGDZRF',
|
||||
'idgreco': '538593',
|
||||
'etat': '1 - Créée',
|
||||
'application': '69999',
|
||||
},
|
||||
)
|
||||
resp = app.post_json(url, params=CREATE_PAYLOAD)
|
||||
create_request = rsps.calls[-1].request
|
||||
|
||||
decoder = MultipartDecoder(create_request.body, create_request.headers['content-type'])
|
||||
assert len(decoder.parts) == 1
|
||||
json_part = decoder.parts[0]
|
||||
assert json_part.headers[b'Content-Type'] == b'application/json'
|
||||
demande_creation = json.loads(json_part.content)
|
||||
assert demande_creation['application'] == '69999'
|
||||
assert demande_creation['localisation']['xgeoloc'] == '50.89491'
|
||||
assert demande_creation['localisation']['ygeoloc'] == '4.34151'
|
||||
assert demande_creation['transmetteur']['nom'] == 'Vertommen'
|
||||
assert demande_creation['transmetteur']['prenom'] == 'Agent 15'
|
||||
assert demande_creation['danger'] == 'false'
|
||||
|
||||
assert not resp.json['err']
|
||||
assert resp.json['data'] == {
|
||||
'iddemande': 'MDFGDZRF',
|
||||
'idgreco': '538593',
|
||||
'etat': '1 - Créée',
|
||||
'application': '69999',
|
||||
}
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, CREATE_OK))
|
||||
def test_greco_create_ok_no_application(mocked_post, app, conn):
|
||||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'create', 'slug': conn.slug})
|
||||
|
@ -271,6 +328,59 @@ def test_greco_create_enclosed(mocked_post, app, conn):
|
|||
assert not resp.json['err']
|
||||
|
||||
|
||||
def test_greco_rest_create_enclosed(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'create', 'slug': rest_conn.slug}
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
|
||||
payload = CREATE_PAYLOAD
|
||||
payload['iddemande'] = ('MDFGDARF',)
|
||||
payload['piece_jointe1'] = {
|
||||
'content': force_str(b64encode(b'my picture content')),
|
||||
'filename': 'foo.png',
|
||||
'content_type': 'image/png',
|
||||
}
|
||||
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.post(
|
||||
'http://greco.example.net/creer',
|
||||
status=200,
|
||||
json={
|
||||
'iddemande': 'MDFGDZRF',
|
||||
'idgreco': '538593',
|
||||
'etat': '1 - Créée',
|
||||
'application': '69999',
|
||||
},
|
||||
)
|
||||
resp = app.post_json(url, params=payload)
|
||||
create_request = rsps.calls[-1].request
|
||||
|
||||
assert create_request.headers['content-type'].startswith('multipart/form-data')
|
||||
decoder = MultipartDecoder(create_request.body, create_request.headers['content-type'])
|
||||
assert len(decoder.parts) == 2
|
||||
|
||||
file_part = decoder.parts[0]
|
||||
assert file_part.headers[b'Content-Type'] == b'image/png'
|
||||
assert file_part.content == b'my picture content'
|
||||
|
||||
json_part = decoder.parts[1]
|
||||
assert json_part.headers[b'Content-Type'] == b'application/json'
|
||||
demande_creation = json.loads(json_part.content)
|
||||
assert demande_creation['application'] == '69999'
|
||||
assert demande_creation['localisation']['xgeoloc'] == '50.89491'
|
||||
assert demande_creation['localisation']['ygeoloc'] == '4.34151'
|
||||
assert demande_creation['transmetteur']['nom'] == 'Vertommen'
|
||||
assert demande_creation['transmetteur']['prenom'] == 'Agent 15'
|
||||
|
||||
assert not resp.json['err']
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, STATUS_OK))
|
||||
def test_greco_status_ok(mocked_post, app, conn):
|
||||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'status', 'slug': conn.slug})
|
||||
|
@ -294,6 +404,54 @@ def test_greco_status_ok(mocked_post, app, conn):
|
|||
}
|
||||
|
||||
|
||||
def test_greco_rest_status_ok(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'status', 'slug': rest_conn.slug}
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.get(
|
||||
'http://greco.example.net/consulter?iddemande=MDFGDZRF&idgreco=538593',
|
||||
status=200,
|
||||
json={
|
||||
'iddemande': 'MDFGDZRF',
|
||||
'idgreco': '538593',
|
||||
'motifsrejet': None,
|
||||
'etat': '1 - Créée',
|
||||
'commentaireReponse': None,
|
||||
'comptesRendus': None,
|
||||
'dateResolutionEffective': None,
|
||||
'dateResolutionPrevue': '2021-03-16T14:14:23Z',
|
||||
'groupeResponsable': 'DRTU/SPS/CIC-Centre info et contact',
|
||||
'suiteDonnee': None,
|
||||
'idOrganismeExterne': None,
|
||||
'nomOrganisme': None,
|
||||
},
|
||||
)
|
||||
resp = app.get(url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF'})
|
||||
|
||||
assert not resp.json['err']
|
||||
assert resp.json['data'] == {
|
||||
'iddemande': 'MDFGDZRF',
|
||||
'idgreco': '538593',
|
||||
'motifsrejet': None,
|
||||
'etat': '1 - Créée',
|
||||
'commentaireReponse': None,
|
||||
'comptesRendus': None,
|
||||
'dateResolutionEffective': None,
|
||||
'dateResolutionPrevue': '2021-03-16T14:14:23Z',
|
||||
'groupeResponsable': 'DRTU/SPS/CIC-Centre info et contact',
|
||||
'suiteDonnee': None,
|
||||
'idOrganismeExterne': None,
|
||||
'nomOrganisme': None,
|
||||
}
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, STATUS_KO))
|
||||
def test_greco_status_ko(mocked_post, app, conn):
|
||||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'status', 'slug': conn.slug})
|
||||
|
@ -343,6 +501,57 @@ def test_greco_answer_ok(mocked_post, app, conn):
|
|||
}
|
||||
|
||||
|
||||
def test_greco_rest_answer_ok(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'answer', 'slug': rest_conn.slug}
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.get(
|
||||
'http://greco.example.net/getMail?iddemande=MDFGDZRF&idgreco=538593',
|
||||
status=200,
|
||||
json={
|
||||
'iddemande': 'DWPQGCTS',
|
||||
'idgreco': '538554',
|
||||
'motifsrejet': None,
|
||||
'etat': None,
|
||||
'message': {
|
||||
'bcc': {'item': []},
|
||||
'cc': {'item': []},
|
||||
'content': '...',
|
||||
'date': '10/02/2021 10:27:47',
|
||||
'subject': 'Métropole de Lyon : réponse à votre demande n° 538554',
|
||||
'to': {'item': ['gbaud@grandlyon.com']},
|
||||
},
|
||||
},
|
||||
)
|
||||
resp = app.get(url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF'})
|
||||
|
||||
assert not resp.json['err']
|
||||
data = resp.json['data']
|
||||
data['message']['content'] = '...'
|
||||
assert data == {
|
||||
'iddemande': 'DWPQGCTS',
|
||||
'idgreco': '538554',
|
||||
'motifsrejet': None,
|
||||
'etat': None,
|
||||
'message': {
|
||||
'bcc': {'item': []},
|
||||
'cc': {'item': []},
|
||||
'content': '...',
|
||||
'date': '10/02/2021 10:27:47',
|
||||
'subject': 'Métropole de Lyon : réponse à votre demande n° 538554',
|
||||
'to': {'item': ['gbaud@grandlyon.com']},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, ANSWER_KO))
|
||||
def test_greco_answer_ko(mocked_post, app, conn):
|
||||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'answer', 'slug': conn.slug})
|
||||
|
@ -375,6 +584,38 @@ def test_greco_add_information_ok(mocked_post, app, conn):
|
|||
}
|
||||
|
||||
|
||||
def test_greco_rest_add_information_ok(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint',
|
||||
kwargs={'connector': 'greco', 'endpoint': 'add-information', 'slug': rest_conn.slug},
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.post(
|
||||
'http://greco.example.net/ajouterComplementInformation',
|
||||
status=200,
|
||||
json={
|
||||
'iddemande': 'ZHRNMWVP',
|
||||
'idgreco': '538634',
|
||||
'motifsrejet': None,
|
||||
},
|
||||
)
|
||||
resp = app.get(url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF', 'information': 'my info'})
|
||||
|
||||
assert not resp.json['err']
|
||||
assert resp.json['data'] == {
|
||||
'iddemande': 'ZHRNMWVP',
|
||||
'idgreco': '538634',
|
||||
'motifsrejet': None,
|
||||
}
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, ADD_INFORMATION_KO))
|
||||
def test_greco_add_information_ko(mocked_post, app, conn):
|
||||
url = reverse(
|
||||
|
@ -405,6 +646,37 @@ def test_greco_update_ok(mocked_post, app, conn):
|
|||
}
|
||||
|
||||
|
||||
def test_greco_rest_update_ok(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'update', 'slug': rest_conn.slug}
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.get(
|
||||
'http://greco.example.net/relancer?iddemande=KPZDXCLL&idgreco=538640&commentaire=mycomment',
|
||||
status=200,
|
||||
json={
|
||||
'iddemande': 'KPZDXCLL',
|
||||
'idgreco': '538640',
|
||||
'motifsrejet': None,
|
||||
},
|
||||
)
|
||||
|
||||
resp = app.get(url, params={'idgreco': '538640', 'iddemande': 'KPZDXCLL', 'comment': 'mycomment'})
|
||||
|
||||
assert not resp.json['err']
|
||||
assert resp.json['data'] == {
|
||||
'iddemande': 'KPZDXCLL',
|
||||
'idgreco': '538640',
|
||||
'motifsrejet': None,
|
||||
}
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, UPDATE_KO))
|
||||
def test_greco_update_ko(mocked_post, app, conn):
|
||||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'update', 'slug': conn.slug})
|
||||
|
@ -435,6 +707,40 @@ def test_greco_add_confirmation_ok(mocked_post, app, conn):
|
|||
}
|
||||
|
||||
|
||||
def test_greco_rest_add_confirmation_ok(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint',
|
||||
kwargs={'connector': 'greco', 'endpoint': 'add-confirmation', 'slug': rest_conn.slug},
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.get(
|
||||
'http://greco.example.net/confirmer?idDemande=MDFGDZRF&idGreco=538593&nbconfirmation=2',
|
||||
status=200,
|
||||
json={
|
||||
'iddemande': 'MDFGDZRF',
|
||||
'idgreco': '538593',
|
||||
'motifsrejet': None,
|
||||
},
|
||||
)
|
||||
resp = app.post_json(
|
||||
url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF', 'nbconfirmation': '2'}
|
||||
)
|
||||
|
||||
assert not resp.json['err']
|
||||
assert resp.json['data'] == {
|
||||
'iddemande': 'MDFGDZRF',
|
||||
'idgreco': '538593',
|
||||
'motifsrejet': None,
|
||||
}
|
||||
|
||||
|
||||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, ADD_CONFIRMATION_KO))
|
||||
def test_greco_add_confirmation_ko(mocked_post, app, conn):
|
||||
url = reverse(
|
||||
|
@ -451,3 +757,49 @@ def test_greco_add_confirmation_ko(mocked_post, app, conn):
|
|||
'idgreco': None,
|
||||
'motifsrejet': 'Le nombre de confirmation doit être un entier\\r\\n',
|
||||
}
|
||||
|
||||
|
||||
def test_greco_rest_no_json_err(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'status', 'slug': rest_conn.slug}
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.get(
|
||||
'http://greco.example.net/consulter?iddemande=MDFGDZRF&idgreco=538593',
|
||||
status=200,
|
||||
content_type='text/html; charset=utf-8',
|
||||
body='some error',
|
||||
)
|
||||
resp = app.get(url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF'})
|
||||
|
||||
assert resp.json['err'] == 1
|
||||
assert resp.json['err_desc'] == 'some error'
|
||||
|
||||
|
||||
def test_greco_rest_bad_json_err(app, rest_conn):
|
||||
url = reverse(
|
||||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'ping', 'slug': rest_conn.slug}
|
||||
)
|
||||
url += '?apikey=grecokey'
|
||||
with responses.RequestsMock() as rsps:
|
||||
rsps.post(
|
||||
'http://greco.example.net/token',
|
||||
status=200,
|
||||
body=get_json_file('token'),
|
||||
)
|
||||
rsps.get(
|
||||
'http://greco.example.net/communicationTest/ping',
|
||||
status=200,
|
||||
content_type='application/json;charset=UTF-8',
|
||||
body='{reponse:"Test de communication ok : ping"}',
|
||||
)
|
||||
resp = app.get(url)
|
||||
|
||||
assert resp.json['err'] == 1
|
||||
assert resp.json['data']['response_text'] == '{reponse:"Test de communication ok : ping"}'
|
||||
|
|
Loading…
Reference in New Issue