2021-03-02 12:56:43 +01:00
|
|
|
|
# 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/>.
|
|
|
|
|
|
|
|
|
|
from base64 import b64encode
|
2020-03-24 15:50:58 +01:00
|
|
|
|
import os
|
|
|
|
|
import pytest
|
|
|
|
|
import mock
|
|
|
|
|
import utils
|
|
|
|
|
|
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
2021-03-02 12:56:43 +01:00
|
|
|
|
from django.urls import reverse
|
|
|
|
|
from django.utils.encoding import force_str
|
2020-03-24 15:50:58 +01:00
|
|
|
|
|
|
|
|
|
from passerelle.base.models import ApiUser, AccessRight
|
|
|
|
|
from passerelle.contrib.greco.models import Greco
|
|
|
|
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
|
|
|
|
|
2021-03-02 12:56:43 +01:00
|
|
|
|
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 utils.FakedResponse(
|
|
|
|
|
content=get_xml_file(basename),
|
|
|
|
|
status_code=200,
|
|
|
|
|
headers={'Content-Type': 'text/xml'},
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
TOKEN = utils.FakedResponse(
|
|
|
|
|
content=get_json_file('token'),
|
|
|
|
|
status_code=200,
|
|
|
|
|
headers={'Content-Type': 'application/json'},
|
|
|
|
|
)
|
|
|
|
|
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')
|
2021-03-05 17:18:26 +01:00
|
|
|
|
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')
|
2021-03-02 12:56:43 +01:00
|
|
|
|
|
|
|
|
|
CREATE_PAYLOAD = {
|
|
|
|
|
'datecreation': '2021-03-02T14:44:38',
|
|
|
|
|
'datedepot': '2021-03-02T14:44:38',
|
|
|
|
|
'domaineobjettype': '1426',
|
|
|
|
|
'iddemande': 'MDFGDZRF',
|
|
|
|
|
'mediareponse': 'Mail',
|
|
|
|
|
'priorite': 'Normal',
|
|
|
|
|
'transmetteur_service': 'GNM',
|
|
|
|
|
'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,
|
|
|
|
|
'beneficiaire_voie': None,
|
|
|
|
|
'beneficiaire_complement': None,
|
|
|
|
|
'beneficiaire_codepostal': None,
|
|
|
|
|
'beneficiaire_commune': None,
|
|
|
|
|
'localisation_numerovoie': '48',
|
|
|
|
|
'localisation_voie': 'Cours Gambetta',
|
|
|
|
|
'localisation_codefuvvoie': '27679',
|
|
|
|
|
'localisation_commune': 'lyon 7',
|
|
|
|
|
'localisation_complement': None,
|
|
|
|
|
}
|
2020-03-24 15:50:58 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def post_effect_503(url, *args, **kwargs):
|
|
|
|
|
if url == 'http://greco.example.net/token':
|
|
|
|
|
return utils.FakedResponse(
|
|
|
|
|
content='{"token_type": "tt", "access_token": "at", "expires_in": "600"}', status_code=200
|
|
|
|
|
)
|
|
|
|
|
return utils.FakedResponse(content='Service Unavailable', status_code=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})
|
|
|
|
|
app.get(url, status=403)
|
|
|
|
|
assert mocked_post.call_count == 0
|
|
|
|
|
|
|
|
|
|
url += '?apikey=grecokey'
|
|
|
|
|
resp = app.get(url).json
|
|
|
|
|
assert mocked_post.call_count == 2 # token + communcationTest (ping -> 503)
|
|
|
|
|
assert resp['err'] == 'transport-error-503'
|
2021-03-02 12:56:43 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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 not resp.json['err']
|
|
|
|
|
assert resp.json['data'] == {
|
|
|
|
|
'iddemande': 'MDFGDZRF',
|
|
|
|
|
'idgreco': '538593',
|
|
|
|
|
'motifsrejet': None,
|
|
|
|
|
'etat': '1 - Créée',
|
|
|
|
|
'application': '69999',
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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']
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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']},
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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},
|
|
|
|
|
}
|
2021-03-05 17:18:26 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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",
|
|
|
|
|
}
|