wip/73841-greco-rest (#73841) #69

Merged
ecazenave merged 2 commits from wip/73841-greco-rest into main 2023-03-14 15:29:17 +01:00
6 changed files with 553 additions and 58 deletions

View File

@ -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 = []

View File

@ -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'),

View File

@ -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',
),
),
]

View File

@ -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}

View File

@ -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"}'

View File

@ -44,6 +44,7 @@ deps =
phonenumbers
dnspython
pyquery
requests_toolbelt
responses
zeep<3.3
codestyle: pre-commit