640 lines
28 KiB
Python
640 lines
28 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import uuid
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import mock
|
|
import pytest
|
|
from dateutil import parser
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from django.utils.encoding import force_bytes
|
|
from requests import Response
|
|
from requests.exceptions import ConnectionError
|
|
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
from passerelle.contrib.iparapheur.models import IParapheur
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
BASE_URL = 'https://secure-iparapheur.demonstrations.adullact.org:443/ws-iparapheur'
|
|
WSDL_URL = '%s?wsdl' % BASE_URL
|
|
API_KEY = 'iparapheur'
|
|
SOAP_NAMESPACES = {
|
|
'soap': 'http://schemas.xmlsoap.org/soap/envelope/',
|
|
'ns1': 'http://www.adullact.org/spring-ws/iparapheur/1.0',
|
|
'xmlmime': 'http://www.w3.org/2005/05/xmlmime',
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def conn():
|
|
api = ApiUser.objects.create(username='iparapheur', keytype='API', key=API_KEY)
|
|
conn = IParapheur.objects.create(
|
|
title='parapheur',
|
|
slug='parapheur',
|
|
wsdl_url=WSDL_URL,
|
|
basic_auth_username='test',
|
|
basic_auth_password='secret',
|
|
)
|
|
obj_type = ContentType.objects.get_for_model(IParapheur)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=conn.pk
|
|
)
|
|
return conn
|
|
|
|
|
|
def assert_invalid_xml(err_desc):
|
|
assert "Server returned response (200) with invalid XML" in err_desc
|
|
|
|
|
|
def xmlmime():
|
|
return os.path.join(os.path.dirname(__file__), 'data', 'xmlmime.xml')
|
|
|
|
|
|
def wsdl_file():
|
|
return os.path.join(os.path.dirname(__file__), 'data', 'iparapheur.wsdl')
|
|
|
|
|
|
def iph_mocked_get(url, params=None, **kwargs):
|
|
response = Response()
|
|
|
|
# simulate first GET call (get wsdl file)
|
|
if url == 'https://secure-iparapheur.demonstrations.adullact.org:443/ws-iparapheur?wsdl':
|
|
target_file = wsdl_file()
|
|
|
|
# simulate second GET call (nested <xsd:import> into wsdl file)
|
|
elif url == 'http://www.w3.org/2005/05/xmlmime':
|
|
assert kwargs['auth'] is None # no basic_authenticton on wsdl' imports
|
|
target_file = xmlmime()
|
|
|
|
else:
|
|
raise Exception('my fault error, url is not yet mocked: %s' % url)
|
|
|
|
with open(target_file, 'rb') as fd:
|
|
response._content = fd.read()
|
|
response.status_code = 200
|
|
return response
|
|
|
|
|
|
@mock.patch('passerelle.contrib.iparapheur.models.IParapheur.get_client')
|
|
def test_call_ping(soap_client, app, conn):
|
|
service = mock.Mock()
|
|
service.echo.return_value = 'pong'
|
|
|
|
class MockedSettings:
|
|
def __init__(self, **kwargs):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
pass
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
pass
|
|
|
|
mocked_client = mock.Mock(overridden_service=service, settings=MockedSettings)
|
|
soap_client.return_value = mocked_client
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'ping', 'slug': conn.slug}
|
|
)
|
|
resp = app.get(url, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == 'pong'
|
|
|
|
|
|
@mock.patch('passerelle.base.models.BaseResource.soap_client', side_effect=ConnectionError('mocked error'))
|
|
def test_call_ping_connectionerror(soap_client, app, conn):
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'ping', 'slug': conn.slug}
|
|
)
|
|
resp = app.get(url, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['data'] is None
|
|
assert 'mocked error' in resp.json['err_desc']
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_create_file(mocked_post, mocked_get, app, conn):
|
|
file_id = str(uuid.uuid4())
|
|
typ = 'Courrier'
|
|
subtyp = 'maire'
|
|
visibility = 'SERVICE'
|
|
email = 'whatever@example.invalid'
|
|
response = Response()
|
|
response.status_code = 200
|
|
|
|
for filename in ('iparapheur_test.odt', 'iparapheur_test.pdf'):
|
|
soap_response = (
|
|
"""<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><CreerDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><MessageRetour><codeRetour>OK</codeRetour><message>Dossier %s soumis dans le circuit</message><severite>INFO</severite></MessageRetour><DossierID>%s</DossierID></CreerDossierResponse></S:Body></S:Envelope>"""
|
|
% (filename, file_id)
|
|
)
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
title, _ = filename.split('.')
|
|
base64_data = 'VGVzdCBEb2N1bWVudA=='
|
|
data = {
|
|
'type': typ,
|
|
'subtype': subtyp,
|
|
'visibility': visibility,
|
|
'title': title,
|
|
'file': {'content': base64_data, 'content_type': 'application/pdf'},
|
|
}
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'iparapheur', 'endpoint': 'create-file', 'slug': conn.slug},
|
|
)
|
|
resp = app.post_json(url, params=data, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.post_json(url, params=data)
|
|
# check output call args
|
|
assert (BASE_URL,) == mocked_post.call_args[0]
|
|
xml = ET.fromstring(mocked_post.call_args[1].get('data'))
|
|
req = xml.find('soap:Body', SOAP_NAMESPACES).find('ns1:CreerDossierRequest', SOAP_NAMESPACES)
|
|
assert req.find('ns1:DossierTitre', SOAP_NAMESPACES).text == title
|
|
assert req.find('ns1:DossierID', SOAP_NAMESPACES).text == title
|
|
assert req.find('ns1:TypeTechnique', SOAP_NAMESPACES).text == typ
|
|
assert req.find('ns1:SousType', SOAP_NAMESPACES).text == subtyp
|
|
assert req.find('ns1:Visibilite', SOAP_NAMESPACES).text == visibility
|
|
assert req.find('ns1:DocumentPrincipal', SOAP_NAMESPACES).text == base64_data
|
|
assert resp.json['data']['dossier_id'] == file_id
|
|
|
|
# Missing dossier_id in response
|
|
title = 'foo'
|
|
soap_response = (
|
|
"""<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><CreerDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><MessageRetour><codeRetour>OK</codeRetour><message>Dossier %s soumis dans le circuit</message><severite>INFO</severite></MessageRetour><DossierID></DossierID></CreerDossierResponse></S:Body></S:Envelope>"""
|
|
% title
|
|
)
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
base64_data = 'VGVzdCBEb2N1bWVudA=='
|
|
data = {
|
|
'type': typ,
|
|
'subtype': subtyp,
|
|
'visibility': visibility,
|
|
'title': title,
|
|
'file': {'content': base64_data, 'content_type': 'application/pdf'},
|
|
}
|
|
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'create-file', 'slug': conn.slug}
|
|
)
|
|
resp = app.post_json(url, params=data, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.post_json(url, params=data)
|
|
assert resp.json['data']['dossier_id'] == title
|
|
|
|
# KO
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><CreerDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><MessageRetour><codeRetour>KO</codeRetour><message>KOmessage</message><severite>INFO</severite></MessageRetour></CreerDossierResponse></S:Body></S:Envelope>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.post_json(url, params=data)
|
|
assert (BASE_URL,) == mocked_post.call_args[0]
|
|
assert resp.json['err'] == 1
|
|
assert 'FileError' in resp.json['err_class']
|
|
assert resp.json['err_desc'] == 'KOmessage'
|
|
|
|
# unknown response
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><CreerDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"></CreerDossierResponse></S:Body></S:Envelope>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.post_json(url, params=data)
|
|
assert (BASE_URL,) == mocked_post.call_args[0]
|
|
assert resp.json['err'] == 1
|
|
assert 'FileError' in resp.json['err_class']
|
|
assert resp.json['err_desc'] == 'unknown error, no response'
|
|
|
|
# bad response
|
|
soap_response = """<nada>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.post_json(url, params=data)
|
|
assert (BASE_URL,) == mocked_post.call_args[0]
|
|
assert resp.json['err'] == 1
|
|
assert 'Transport Error' in resp.json['err_desc']
|
|
assert_invalid_xml(resp.json['err_desc'])
|
|
|
|
# Unknown value for "visibility"
|
|
err_data = data.copy()
|
|
err_data['visibility'] = 'UNKNOWN_VISIBILITY'
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'create-file', 'slug': conn.slug}
|
|
)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.post_json(url, params=err_data)
|
|
assert resp.json['err'] == 1
|
|
assert 'FileError' in resp.json['err_class']
|
|
|
|
# OK, providing email
|
|
data['email'] = email
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'create-file', 'slug': conn.slug}
|
|
)
|
|
url += '?apikey=%s' % API_KEY
|
|
soap_response = """<nada>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.post_json(url, params=data)
|
|
assert (BASE_URL,) == mocked_post.call_args[0]
|
|
assert resp.json['err'] == 1
|
|
assert 'Transport Error' in resp.json['err_desc']
|
|
assert_invalid_xml(resp.json['err_desc'])
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_files(mocked_post, mocked_get, app, conn):
|
|
response = Response()
|
|
response.status_code = 200
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><RechercherDossiersResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><LogDossier><timestamp>2015-10-29T15:42:08.732+01:00</timestamp><nom>test</nom><status>Lu</status><annotation></annotation><accessible>KO</accessible></LogDossier><LogDossier><timestamp>2015-10-29T15:52:42.167+01:00</timestamp><nom>Test2</nom><status>RejetSignataire</status><annotation></annotation><accessible>OK</accessible></LogDossier><LogDossier><timestamp>2015-11-25T12:13:30.830+01:00</timestamp><nom>6ceecfb7-67ee-4388-8943-35911c640031</nom><status>NonLu</status><annotation></annotation><accessible>Recuperable</accessible></LogDossier></RechercherDossiersResponse></S:Body></S:Envelope>
|
|
"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'files', 'slug': conn.slug}
|
|
)
|
|
resp = app.get(url, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert len(resp.json['data']) == 3
|
|
for item in resp.json['data']:
|
|
assert item['status']
|
|
assert item['id']
|
|
assert item['timestamp']
|
|
|
|
# same call providing parameter
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><RechercherDossiersResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><LogDossier><timestamp>2015-10-29T15:42:08.732+01:00</timestamp><nom>test</nom><status>Lu</status><annotation></annotation><accessible>KO</accessible></LogDossier></RechercherDossiersResponse></S:Body></S:Envelope>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
url += '&status=Lu'
|
|
resp = app.get(url)
|
|
assert len(resp.json['data']) == 1
|
|
for item in resp.json['data']:
|
|
assert item['status']
|
|
assert item['id']
|
|
assert item['timestamp']
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_get_file_status(mocked_post, mocked_get, app, conn):
|
|
file_id = str(uuid.uuid4())
|
|
response = Response()
|
|
response.status_code = 200
|
|
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><GetHistoDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><LogDossier><timestamp>2016-04-05T17:57:03.893+02:00</timestamp><nom>webservices gru</nom><status>NonLu</status><annotation>Création de dossier</annotation></LogDossier><LogDossier><timestamp>2016-04-05T17:58:46.716+02:00</timestamp><nom>webservices gru</nom><status>NonLu</status><annotation>super</annotation></LogDossier><LogDossier><timestamp>2016-04-05T17:58:46.727+02:00</timestamp><nom>webservices gru</nom><status>Archive</status><annotation>Circuit terminé, dossier archivable</annotation></LogDossier><MessageRetour><codeRetour>OK</codeRetour><message></message><severite>INFO</severite></MessageRetour></GetHistoDossierResponse></S:Body></S:Envelope>
|
|
"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'connector': 'iparapheur', 'endpoint': 'get-file-status', 'slug': conn.slug, 'rest': file_id},
|
|
)
|
|
resp = app.get(url, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert len(resp.json) == 2
|
|
data = resp.json['data']
|
|
assert data['status'] == 'Archive'
|
|
assert data['nom'] == 'webservices gru'
|
|
expected_ts = timezone.make_naive(
|
|
parser.parse('2016-04-05T17:58:46.727+02:00'), timezone.get_current_timezone()
|
|
)
|
|
assert data['timestamp'] in (
|
|
expected_ts.strftime('%Y-%m-%dT%H:%M:%S.000'), # suds 0.4
|
|
'2016-04-05T17:58:46.727+02:00', # suds 0.7
|
|
)
|
|
|
|
# KO
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><CreerDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><MessageRetour><codeRetour>KO</codeRetour><message>KOmessage</message><severite>INFO</severite></MessageRetour></CreerDossierResponse></S:Body></S:Envelope>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert 'FileError' in resp.json['err_class']
|
|
assert resp.json['err_desc'] == 'KOmessage'
|
|
|
|
# unknown response
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><CreerDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"></CreerDossierResponse></S:Body></S:Envelope>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert 'FileError' in resp.json['err_class']
|
|
assert resp.json['err_desc'] == 'unknown error, no response'
|
|
|
|
# bad response
|
|
soap_response = """<nada>"""
|
|
response._content = soap_response
|
|
mocked_post.return_value = response
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert 'passerelle.utils.jsonresponse.APIError' in resp.json['err_class']
|
|
assert_invalid_xml(resp.json['err_desc'])
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_get_file(mocked_post, mocked_get, app, conn):
|
|
file_id = str(uuid.uuid4())
|
|
response = Response()
|
|
response.status_code = 200
|
|
|
|
with open(os.path.join(os.path.dirname(__file__), 'data/iparapheur_get_file_response.xml'), 'rb') as fd:
|
|
soap_response = fd.read()
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'get-file', 'rest': file_id},
|
|
)
|
|
resp = app.get(url, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert resp.headers['Content-Type'] == 'application/pdf'
|
|
assert 'filename=test.pdf' in resp.headers['Content-Disposition']
|
|
assert resp.text == 'Test Document'
|
|
|
|
# KO
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><GetDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><MessageRetour><codeRetour>KO</codeRetour><message>KOmessage</message><severite>INFO</severite></MessageRetour></GetDossierResponse></S:Body></S:Envelope>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert 'FileError' in resp.json['err_class']
|
|
assert resp.json['err_desc'] == 'KOmessage'
|
|
|
|
# unknown response
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><GetDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"></GetDossierResponse></S:Body></S:Envelope>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert 'FileError' in resp.json['err_class']
|
|
assert resp.json['err_desc'] == 'unknown error, no response'
|
|
|
|
# bad response
|
|
soap_response = """<nada>"""
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert 'passerelle.utils.jsonresponse.APIError' in resp.json['err_class']
|
|
assert 'XMLSyntax Error' in resp.json['err_desc']
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_get_file_invalid_appendix(mocked_post, mocked_get, app, conn):
|
|
file_id = str(uuid.uuid4())
|
|
response = Response()
|
|
response.status_code = 200
|
|
|
|
with open(os.path.join(os.path.dirname(__file__), 'data/iparapheur_get_file_response.xml'), 'rb') as fd:
|
|
soap_response = fd.read()
|
|
response._content = soap_response
|
|
mocked_post.return_value = response
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'get-file', 'rest': file_id},
|
|
)
|
|
url += '?apikey=%s&appendix=invalid' % API_KEY
|
|
resp = app.get(url, status=404)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'invalid appendix index'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_get_file_not_found_appendix(mocked_post, mocked_get, app, conn):
|
|
file_id = str(uuid.uuid4())
|
|
response = Response()
|
|
response.status_code = 200
|
|
|
|
with open(os.path.join(os.path.dirname(__file__), 'data/iparapheur_get_file_response.xml'), 'rb') as fd:
|
|
soap_response = fd.read()
|
|
response._content = soap_response
|
|
mocked_post.return_value = response
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'get-file', 'rest': file_id},
|
|
)
|
|
url += '?apikey=%s&appendix=10' % API_KEY
|
|
resp = app.get(url, status=404)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'unknown appendix'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_get_file_appendix(mocked_post, mocked_get, app, conn):
|
|
file_id = str(uuid.uuid4())
|
|
response = Response()
|
|
response.status_code = 200
|
|
|
|
with open(os.path.join(os.path.dirname(__file__), 'data/iparapheur_get_file_response.xml'), 'rb') as fd:
|
|
soap_response = fd.read()
|
|
response._content = soap_response
|
|
mocked_post.return_value = response
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'get-file', 'rest': file_id},
|
|
)
|
|
url += '?apikey=%s&appendix=0' % API_KEY
|
|
resp = app.get(url)
|
|
assert resp.headers['Content-Type'] == 'application/pdf'
|
|
assert 'filename=iParapheur_impression_dossier.pdf' in resp.headers['Content-Disposition']
|
|
assert 'filename*=UTF-8\'\'iParapheur_impression_dossier.pdf' in resp.headers['Content-Disposition']
|
|
assert resp.text == 'Test Document Appendix'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_invalid_response(mocked_post, mocked_get, app, conn):
|
|
file_id = str(uuid.uuid4())
|
|
response = Response()
|
|
response.status_code = 502
|
|
|
|
response._content = '<p>Bad Gateway</p>'
|
|
response.raison = 'Bad Gateway'
|
|
mocked_post.return_value = response
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'get-file-status', 'rest': file_id},
|
|
)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert resp.json['err_desc'] == "ServiceError: Unknown fault occured"
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_webfault_response(mocked_post, mocked_get, app, conn):
|
|
file_id = str(uuid.uuid4())
|
|
response = Response()
|
|
response.status_code = 200
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'get-file-status', 'rest': file_id},
|
|
)
|
|
url += '?apikey=%s' % API_KEY
|
|
|
|
webfault_response = """<?xml version='1.0' encoding='UTF-8'?>
|
|
<SOAP-ENV:Envelope xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/envelope/" xmlns:xsi="http://www.w3.org/1999/XMLSchema-instance" xmlns:xsd="http://www.w3.org/1999/XMLSchema">
|
|
<SOAP-ENV:Body>
|
|
<SOAP-ENV:Fault>
|
|
<faultcode xsi:type="xsd:string">SOAP-ENV:Client</faultcode>
|
|
<faultstring xsi:type="xsd:string">
|
|
Test server error
|
|
</faultstring>
|
|
</SOAP-ENV:Fault>
|
|
</SOAP-ENV:Body>
|
|
</SOAP-ENV:Envelope>"""
|
|
response._content = force_bytes(webfault_response)
|
|
mocked_post.return_value = response
|
|
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert 'ServiceError:' in resp.json['err_desc']
|
|
assert 'Test server error' in resp.json['err_desc']
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
def test_call_wsdl(mocked_get, app, conn):
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'wsdl'}
|
|
)
|
|
resp = app.get(url)
|
|
assert resp.headers['content-type'] == 'text/xml'
|
|
with open(wsdl_file(), 'rb') as fd:
|
|
assert resp.content == fd.read()
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=ConnectionError('mocked error'))
|
|
def test_call_wsdl_connectionerror(mocked_get, app, conn):
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'wsdl', 'slug': conn.slug}
|
|
)
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['data'] is None
|
|
assert 'mocked error' in resp.json['err_desc']
|
|
|
|
|
|
@pytest.mark.xfail(run=False)
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
@mock.patch('zeep.Transport._load_remote_data')
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_no_auth_on_wsdl_imports(mocked_post, mocked_load, mocked_get, app, conn):
|
|
"""
|
|
While getting wsdl:
|
|
- zeep.Transport._load_remote_data is call on basic_authentication (get wsdl file)
|
|
- passerelle.utils.Request.get is call on no basic_authenticaion (get xmlmime import's file)
|
|
|
|
Notice: this also test the 'echo' SOAP service
|
|
"""
|
|
response_xmlmime, response_post = Response(), Response()
|
|
response_xmlmime.status_code, response_post.status_code = 200, 200
|
|
with open(xmlmime(), 'rb') as fd:
|
|
response_xmlmime._content = fd.read()
|
|
response_post._content = force_bytes(
|
|
"""<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><echoResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime">[publik_test] m'a dit: "ping"!</echoResponse></S:Body></S:Envelope>
|
|
"""
|
|
)
|
|
with open(wsdl_file(), 'rb') as fd:
|
|
mocked_load.return_value = fd.read()
|
|
mocked_get.return_value = response_xmlmime
|
|
mocked_post.return_value = response_post
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'ping', 'slug': conn.slug}
|
|
)
|
|
|
|
resp = app.get(url, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == "[publik_test] m'a dit: \"ping\"!"
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_types(mocked_post, mocked_get, app, conn):
|
|
"""test GetListeTypesRequest SOAP service"""
|
|
response = Response()
|
|
response.status_code = 200
|
|
|
|
# empty list of types as response
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><GetListeTypesResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"/></S:Body></S:Envelope>"""
|
|
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'types'}
|
|
)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == []
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_subtypes(mocked_post, mocked_get, app, conn):
|
|
"""test GetListeSousTypes SOAP service"""
|
|
response = Response()
|
|
response.status_code = 200
|
|
|
|
# error: no parameter provided
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'slug': conn.slug, 'connector': 'iparapheur', 'endpoint': 'subtypes'}
|
|
)
|
|
resp = app.get(url, status=403)
|
|
url += '?apikey=%s' % API_KEY
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 1
|
|
assert (
|
|
resp.json['err_desc']
|
|
== "Type Error: TypeTechnique() takes exactly 1 argument (0 given). Simple types expect only a single value argument"
|
|
)
|
|
|
|
# providing a type as parameter
|
|
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><GetListeTypesResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"/></S:Body></S:Envelope>"""
|
|
|
|
response._content = force_bytes(soap_response) # empty list of subtypes as response
|
|
mocked_post.return_value = response
|
|
url += '&type=my_unknown_type'
|
|
resp = app.get(url)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == []
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
|
|
@mock.patch('passerelle.utils.Request.post')
|
|
def test_call_ping_overrinding_endpoint_url(mocked_post, mocked_get, app, conn):
|
|
response = Response()
|
|
response.status_code = 200
|
|
soap_response = 'no importance'
|
|
response._content = force_bytes(soap_response)
|
|
mocked_post.return_value = response
|
|
|
|
conn.wsdl_endpoint_location = 'http://www.new.location.com'
|
|
conn.save()
|
|
url = reverse(
|
|
'generic-endpoint', kwargs={'connector': 'iparapheur', 'endpoint': 'ping', 'slug': conn.slug}
|
|
)
|
|
url += '?apikey=%s' % API_KEY
|
|
app.get(url)
|
|
assert mocked_post.call_args[0][0] == 'http://www.new.location.com'
|