passerelle/tests/test_iparapheur.py

445 lines
23 KiB
Python

# -*- coding: utf-8 -*-
import pytest
import mock
import uuid
import os
import base64
import xml.etree.ElementTree as ET
from dateutil import parser
from requests.exceptions import ConnectionError
from django.core.urlresolvers import reverse
from django.contrib.contenttypes.models import ContentType
from django.utils import timezone
from passerelle.base.models import ApiUser, AccessRight
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, username='test',
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
@pytest.fixture
def xmlmime():
return os.path.join(os.path.dirname(__file__), 'data','xmlmime.xml')
@pytest.fixture
def wsdl_file():
return os.path.join(os.path.dirname(__file__), 'data','iparapheur.wsdl')
@mock.patch('passerelle.contrib.iparapheur.models.soap_get_client')
def test_call_ping(soap_client, app, conn):
service = mock.Mock()
service.echo.return_value = 'pong'
mocked_client = mock.Mock(service=service)
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.contrib.iparapheur.models.soap_get_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'] == None
assert 'mocked error' in resp.json['err_desc']
@mock.patch('passerelle.utils.Request.get')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_create_file(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
file_id = str(uuid.uuid4())
typ = 'Courrier'
subtyp = 'maire'
visibility = 'SERVICE'
for filename in ('iparapheur_test.odt', 'iparapheur_test.pdf'):
title, ext = filename.split('.')
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>
""" % (title, file_id)
base64_data = 'VGVzdCBEb2N1bWVudA=='
data = {'type': typ, 'subtype': subtyp, 'visibility': visibility,
'title': title, 'data': base64_data, 'content-type':'application/pdf'}
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content = file(wsdl_file).read(),
status_code=200)
mocked_post.return_value = mock.Mock(status_code=200,
content=soap_response)
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: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']['RecordId'] == file_id
# 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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.post_json(url, params=data, status=500)
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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.post_json(url, params=data, status=500)
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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.post_json(url, params=data, status=500)
assert resp.json['err'] == 1
assert 'SAXParseException' in resp.json['err_class']
assert 'syntax error' in resp.json['err_desc']
@mock.patch('passerelle.utils.Request.get')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_files(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
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>
"""
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content = file(wsdl_file).read(),
status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_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']
@mock.patch('passerelle.utils.Request.get')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_get_file_status(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
file_id = str(uuid.uuid4())
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>
"""
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content = file(wsdl_file).read(),
status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.get(url, status=500)
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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.get(url, status=500)
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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.get(url, status=500)
assert resp.json['err'] == 1
assert 'SAXParseException' in resp.json['err_class']
assert 'syntax error' in resp.json['err_desc']
@mock.patch('passerelle.utils.Request.get')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_get_file(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
file_id = str(uuid.uuid4())
soap_response = file(os.path.join(os.path.dirname(__file__),
'data/iparapheur_get_file_response.xml')).read()
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content = file(wsdl_file).read(),
status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_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.body == 'Test Document'
# 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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.get(url, status=500)
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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.get(url, status=500)
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>"""
http_open.return_value = open(xmlmime)
mocked_get.return_value = mock.Mock(content=open(wsdl_file).read(), status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_response)
resp = app.get(url, status=500)
assert resp.json['err'] == 1
assert 'SAXParseException' in resp.json['err_class']
assert 'syntax error' in resp.json['err_desc']
@mock.patch('passerelle.utils.Request.get')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_get_file_invalid_appendix(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
file_id = str(uuid.uuid4())
soap_response = file(os.path.join(os.path.dirname(__file__),
'data/iparapheur_get_file_response.xml')).read()
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content=file(wsdl_file).read(),
status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_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')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_get_file_not_found_appendix(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
file_id = str(uuid.uuid4())
soap_response = file(os.path.join(os.path.dirname(__file__),
'data/iparapheur_get_file_response.xml')).read()
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content=file(wsdl_file).read(),
status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_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')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_get_file_appendix(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
file_id = str(uuid.uuid4())
soap_response = file(os.path.join(os.path.dirname(__file__),
'data/iparapheur_get_file_response.xml')).read()
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content=file(wsdl_file).read(),
status_code=200)
mocked_post.return_value = mock.Mock(status_code=200, content=soap_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.body == 'Test Document Appendix'
@mock.patch('passerelle.utils.Request.get')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_invalid_response(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
file_id = str(uuid.uuid4())
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content=file(wsdl_file).read(),
status_code=200)
mocked_post.return_value = mock.Mock(status_code=502,
content='<p>Bad Gateway</p>', reason='Bad Gateway', ok=False)
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'] == "Transport Error: (502, u'Bad Gateway')"
@mock.patch('passerelle.utils.Request.get')
@mock.patch('passerelle.utils.Request.post')
@mock.patch('passerelle.contrib.iparapheur.soap.HttpAuthenticated.open')
def test_webfault_response(http_open, mocked_post, mocked_get, app, conn, xmlmime, wsdl_file):
file_id = str(uuid.uuid4())
url = reverse('generic-endpoint', kwargs={'slug': conn.slug,
'connector': 'iparapheur', 'endpoint': 'get-file-status', 'rest': file_id})
url += '?apikey=%s' % API_KEY
http_open.return_value = file(xmlmime)
mocked_get.return_value = mock.Mock(content=file(wsdl_file).read(),
status_code=200)
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>"""
mocked_post.return_value = mock.Mock(
status_code=200, content=webfault_response,
ok=True)
resp = app.get(url)
assert 'ServiceError' in resp.json['err_desc']
assert 'Test server error' in resp.json['err_desc']
def test_get_magic_mime(tmpdir):
from PIL import Image
from passerelle.contrib.iparapheur.models import get_magic_mime
image = Image.new("RGB", (10, 10), (255, 255, 255))
image_path = tmpdir.join('image.jpeg')
image.save(image_path.strpath)
with image_path.open() as f:
assert 'image/jpeg' == get_magic_mime(f.read())
text_path = tmpdir.join('text_file.txt')
with text_path.open('w') as f:
f.write('some text')
with text_path.open() as f:
assert 'text/plain' == get_magic_mime(f.read())
@mock.patch('passerelle.utils.Request.get')
def test_call_wsdl(mocked_get, app, conn, wsdl_file):
mocked_get.return_value = mock.Mock(content=file(wsdl_file).read(),
status_code=200)
url = reverse('generic-endpoint',
kwargs={'slug': conn.slug,
'connector': 'iparapheur', 'endpoint': 'wsdl'})
resp = app.get(url)
assert resp.headers['content-type'] == 'text/xml'
assert resp.content == file(wsdl_file).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']