859 lines
30 KiB
Python
859 lines
30 KiB
Python
# Copyright (C) 2016 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 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
|
||
from passerelle.contrib.greco.models import Greco
|
||
|
||
pytestmark = pytest.mark.django_db
|
||
|
||
|
||
TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'greco')
|
||
WSDL_FILENAME = os.path.join(TEST_BASE_DIR, 'greco.wsdl')
|
||
|
||
|
||
def get_xml_file(filename):
|
||
with open(os.path.join(TEST_BASE_DIR, '%s.xml' % filename), 'rb') as desc:
|
||
return desc.read()
|
||
|
||
|
||
def get_json_file(filename):
|
||
with open(os.path.join(TEST_BASE_DIR, '%s.json' % filename)) as desc:
|
||
return desc.read()
|
||
|
||
|
||
def fake_xml_response(basename):
|
||
return tests.utils.FakedResponse(
|
||
content=get_xml_file(basename),
|
||
status_code=200,
|
||
headers={'Content-Type': 'text/xml'},
|
||
)
|
||
|
||
|
||
TOKEN = tests.utils.FakedResponse(
|
||
content=get_json_file('token'),
|
||
status_code=200,
|
||
headers={'Content-Type': 'application/json'},
|
||
)
|
||
RESP_503 = tests.utils.FakedResponse(content='Service Unavailable', status_code=503)
|
||
PING = fake_xml_response('ping')
|
||
CREATE_OK = fake_xml_response('create_ok')
|
||
CREATE_KO = fake_xml_response('create_ko')
|
||
STATUS_OK = fake_xml_response('status_ok')
|
||
STATUS_KO = fake_xml_response('status_ko')
|
||
ANSWER_OK = fake_xml_response('answer_ok')
|
||
ANSWER_KO = fake_xml_response('answer_ko')
|
||
ADD_INFORMATION_OK = fake_xml_response('add_information_ok')
|
||
ADD_INFORMATION_KO = fake_xml_response('add_information_ko')
|
||
UPDATE_OK = fake_xml_response('update_ok')
|
||
UPDATE_KO = fake_xml_response('update_ko')
|
||
ADD_CONFIRMATION_OK = fake_xml_response('add_confirmation_ok')
|
||
ADD_CONFIRMATION_KO = fake_xml_response('add_confirmation_ko')
|
||
|
||
CREATE_PAYLOAD = {
|
||
'datecreation': '2021-03-02T14:44:38',
|
||
'datedepot': '2021-03-02T14:44:38',
|
||
'domaineobjettype': '1426',
|
||
'iddemande': 'MDFGDZRF',
|
||
'mediareponse': 'Mail',
|
||
'priorite': 'Normal',
|
||
'application': '69999',
|
||
'description': 'Nom de l’usager\xa0:\n admin admin\n '
|
||
+ 'Carte\xa0:\n 45.75399700317313;4.846451997582336\nNuméro\xa0:\n '
|
||
+ '48\n Nom de la voie\xa0:\n Cours Gambetta\n Code postal\xa0:\n '
|
||
+ '69007\nCommune\xa0:\n Lyon\nVotre demande concerne :\xa0:\n '
|
||
+ 'Un nouvel espace cyclable, bande, piste...\n'
|
||
+ 'Votre demande\xa0:\n Une nouvelle piste cyclable, svp.\n',
|
||
'beneficiaire_typetiers': 'Particulier',
|
||
'beneficiaire_email': 'jdoe@example.fr',
|
||
'beneficiaire_civilite': 'Monsieur',
|
||
'beneficiaire_prenom': 'Jhon',
|
||
'beneficiaire_nom': 'Doe',
|
||
'beneficiaire_telephone': '',
|
||
'beneficiaire_mobile': '0612345678',
|
||
'beneficiaire_numerovoie': None,
|
||
'localisation_numerovoie': '48',
|
||
'localisation_voie': 'Cours Gambetta',
|
||
'localisation_codefuvvoie': '27679',
|
||
'localisation_commune': 'lyon 7',
|
||
'localisation_xgeoloc': '50.89491',
|
||
'localisation_ygeoloc': '4.34151',
|
||
'transmetteur_service': 'GNM',
|
||
'transmetteur_nom': 'Vertommen',
|
||
'transmetteur_prenom': 'Agent 15',
|
||
'danger': False,
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def conn():
|
||
api_user = ApiUser.objects.create(username='greco', keytype='API', key='grecokey')
|
||
connector = Greco.objects.create(
|
||
title='Greco',
|
||
slug='greco',
|
||
application='appid',
|
||
token_url='http://greco.example.net/token',
|
||
token_authorization='authtoken',
|
||
wsdl_url='file://%s' % WSDL_FILENAME,
|
||
verify_cert=False,
|
||
)
|
||
obj_type = ContentType.objects.get_for_model(Greco)
|
||
AccessRight.objects.create(
|
||
codename='can_access', apiuser=api_user, resource_type=obj_type, resource_pk=connector.pk
|
||
)
|
||
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})
|
||
app.get(url, status=403)
|
||
assert mocked_post.call_count == 0
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(RESP_503,))
|
||
def test_greco_no_token(mocked_post, app, conn):
|
||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'ping', 'slug': conn.slug})
|
||
url += '?apikey=grecokey'
|
||
resp = app.get(url).json
|
||
assert mocked_post.call_count == 1 # token (-> 503)
|
||
assert resp['err'] == 'transport-error-503'
|
||
|
||
|
||
def post_effect_503(url, *args, **kwargs):
|
||
if url == 'http://greco.example.net/token':
|
||
return TOKEN
|
||
return RESP_503
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=post_effect_503)
|
||
def test_greco_ping(mocked_post, app, conn):
|
||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'ping', 'slug': conn.slug})
|
||
url += '?apikey=grecokey'
|
||
resp = app.get(url).json
|
||
assert mocked_post.call_count == 2 # token + communcationTest (ping -> 503)
|
||
assert resp['err'] == 'transport-error-503'
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, PING))
|
||
def test_greco_ping_ok(mocked_post, app, conn):
|
||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'ping', 'slug': conn.slug})
|
||
url += '?apikey=grecokey'
|
||
resp = app.get(url)
|
||
assert not resp.json['err']
|
||
assert (
|
||
resp.json['data']
|
||
== 'com.coheris.external.services.mairies.action.TestCommunicationAction.execute(...)\\n[ping]'
|
||
)
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, CREATE_OK))
|
||
def test_greco_create_ok(mocked_post, app, conn):
|
||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'create', 'slug': conn.slug})
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.post_json(url, params=CREATE_PAYLOAD)
|
||
assert mocked_post.call_count == 2
|
||
|
||
def to_json(root):
|
||
tag = root.tag[root.tag.find('}') + 1 :]
|
||
if len(root) == 0: # text node
|
||
return {tag: root.text or None}
|
||
childs = {}
|
||
for child in root:
|
||
childs.update(to_json(child))
|
||
return {tag: childs or None}
|
||
|
||
call_args = to_json(ET.fromstring(mocked_post.call_args[1]['data']))
|
||
demande_creation = call_args['Envelope']['Body']['creer']['demandeCreation']
|
||
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']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'MDFGDZRF',
|
||
'idgreco': '538593',
|
||
'motifsrejet': None,
|
||
'etat': '1 - Créée',
|
||
'application': '69999',
|
||
}
|
||
|
||
|
||
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['beneficiaire']['fax'] == ''
|
||
assert demande_creation['beneficiaire']['numerovoie'] == ''
|
||
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',
|
||
}
|
||
|
||
|
||
def test_greco_rest_create_ok_default_application_code(app, rest_conn):
|
||
rest_conn.application = '69999'
|
||
rest_conn.save()
|
||
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',
|
||
},
|
||
)
|
||
create_payload = CREATE_PAYLOAD.copy()
|
||
del create_payload['application']
|
||
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['beneficiaire']['fax'] == ''
|
||
assert demande_creation['beneficiaire']['numerovoie'] == ''
|
||
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})
|
||
url += '?apikey=grecokey'
|
||
|
||
payload = copy.copy(CREATE_PAYLOAD)
|
||
del payload['application']
|
||
app.post_json(url, params=payload)
|
||
assert mocked_post.call_count == 2
|
||
|
||
def to_json(root):
|
||
tag = root.tag[root.tag.find('}') + 1 :]
|
||
if len(root) == 0: # text node
|
||
return {tag: root.text or None}
|
||
childs = {}
|
||
for child in root:
|
||
childs.update(to_json(child))
|
||
return {tag: childs or None}
|
||
|
||
call_args = to_json(ET.fromstring(mocked_post.call_args[1]['data']))
|
||
demande_creation = call_args['Envelope']['Body']['creer']['demandeCreation']
|
||
assert demande_creation['application'] == 'appid'
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, CREATE_KO))
|
||
def test_greco_create_ko(mocked_post, app, conn):
|
||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'create', 'slug': conn.slug})
|
||
url += '?apikey=grecokey'
|
||
|
||
payload = CREATE_PAYLOAD
|
||
payload['beneficiaire_email'] = 'jdoe@localhost'
|
||
resp = app.post_json(url, params=payload)
|
||
assert not resp.json['err']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'MDFGDZRF',
|
||
'idgreco': None,
|
||
'motifsrejet': 'Beneficiaire: Adresse mail incorrecte\\r\\n',
|
||
'etat': None,
|
||
'application': '69999',
|
||
}
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, CREATE_OK))
|
||
def test_greco_create_enclosed(mocked_post, app, conn):
|
||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'create', 'slug': 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',
|
||
}
|
||
|
||
resp = app.post_json(url, params=payload)
|
||
|
||
# RFC 2045 MIME multipart boundaries
|
||
assert force_str(mocked_post.call_args[1]['data']).count('==\r\n') == 2
|
||
assert force_str(mocked_post.call_args[1]['data']).count('\r\n--==') == 2
|
||
|
||
# substitute binary parts
|
||
assert 'Content-ID: <foo.png>' in force_str(mocked_post.call_args[1]['data'])
|
||
|
||
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})
|
||
url += '?apikey=grecokey'
|
||
|
||
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,
|
||
}
|
||
|
||
|
||
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})
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.get(url, params={'idgreco': '538999', 'iddemande': 'AAAAAARF'})
|
||
assert not resp.json['err']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'AAAAAARF',
|
||
'idgreco': '538999',
|
||
'motifsrejet': "La demande n\\'existe pas\\r\\n",
|
||
'etat': None,
|
||
'commentaireReponse': None,
|
||
'comptesRendus': None,
|
||
'dateResolutionEffective': None,
|
||
'dateResolutionPrevue': None,
|
||
'groupeResponsable': None,
|
||
'suiteDonnee': None,
|
||
'idOrganismeExterne': None,
|
||
'nomOrganisme': None,
|
||
}
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, ANSWER_OK))
|
||
def test_greco_answer_ok(mocked_post, app, conn):
|
||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'answer', 'slug': conn.slug})
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.get(url, params={'idgreco': '538554', 'iddemande': 'DWPQGCTS'})
|
||
assert not resp.json['err']
|
||
data = resp.json['data']
|
||
assert len(data['message']['content']) == 2433
|
||
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']},
|
||
},
|
||
}
|
||
|
||
|
||
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': '',
|
||
'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})
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.get(url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF'})
|
||
assert not resp.json['err']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'MDFGDZRF',
|
||
'idgreco': '538593',
|
||
'motifsrejet': 'Aucune réponse trouvée',
|
||
'etat': None,
|
||
'message': {'bcc': None, 'cc': None, 'content': None, 'date': None, 'subject': None, 'to': None},
|
||
}
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, ADD_INFORMATION_OK))
|
||
def test_greco_add_information_ok(mocked_post, app, conn):
|
||
url = reverse(
|
||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'add-information', 'slug': conn.slug}
|
||
)
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.get(url, params={'idgreco': '538634', 'iddemande': 'ZHRNMWVP', 'information': 'my info'})
|
||
assert not resp.json['err']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'ZHRNMWVP',
|
||
'idgreco': '538634',
|
||
'motifsrejet': None,
|
||
}
|
||
|
||
|
||
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(
|
||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'add-information', 'slug': conn.slug}
|
||
)
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.get(url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF', 'information': 'my info'})
|
||
assert not resp.json['err']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'MDFGDZRF',
|
||
'idgreco': '538593',
|
||
'motifsrejet': "Le complément d\\'information ne peut être ajouté qu\\'à une demande ayant le statut \\'incomplète\\'\\r\\n",
|
||
}
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, UPDATE_OK))
|
||
def test_greco_update_ok(mocked_post, app, conn):
|
||
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'update', 'slug': conn.slug})
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.get(url, params={'idgreco': '538640', 'iddemande': 'KPZDXCLL', 'comment': 'my comment'})
|
||
assert not resp.json['err']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'KPZDXCLL',
|
||
'idgreco': '538640',
|
||
'motifsrejet': None,
|
||
}
|
||
|
||
|
||
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})
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.get(url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF', 'comment': 'my comment'})
|
||
assert not resp.json['err']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'MDFGDZRF',
|
||
'idgreco': '538593',
|
||
'motifsrejet': "La date de résolution prévue pour la demande 538593 n\\'est pas dépassée\\r\\n",
|
||
}
|
||
|
||
|
||
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, ADD_CONFIRMATION_OK))
|
||
def test_greco_add_confirmation_ok(mocked_post, app, conn):
|
||
url = reverse(
|
||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'add-confirmation', 'slug': conn.slug}
|
||
)
|
||
url += '?apikey=grecokey'
|
||
|
||
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,
|
||
}
|
||
|
||
|
||
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(
|
||
'generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'add-confirmation', 'slug': conn.slug}
|
||
)
|
||
url += '?apikey=grecokey'
|
||
|
||
resp = app.post_json(
|
||
url, params={'idgreco': '538593', 'iddemande': 'MDFGDZRF', 'nbconfirmation': 'a lot'}
|
||
)
|
||
assert not resp.json['err']
|
||
assert resp.json['data'] == {
|
||
'iddemande': 'MDFGDZRF',
|
||
'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"}'
|