0001-iparapheur-get_file-allow-missing-element-in-xsd-seq.patch

Emmanuel Cazenave, 25 juin 2019 17:21

Voir les différences: en ligne côte à côte

Subject: [PATCH] iparapheur: get_file, allow missing element in xsd sequence
 (#34299)

 passerelle/contrib/iparapheur/models.py | 54 +++++++++++++------------
 tests/test_iparapheur.py                | 12 ++----
 2 files changed, 32 insertions(+), 34 deletions(-)

     from django.utils.translation import ugettext_lazy as _
     from django.http import HttpResponse, Http404
     from zeep.exceptions import Fault as WebFault, TransportError
     from zeep.exceptions import Fault as WebFault, TransportError, XMLSyntaxError
     from passerelle.base.models import BaseResource, HTTPResource
     from passerelle.utils.api import endpoint
-...
+    }
     def get_client(model):
         try:
             soap_client = model.soap_client()
             # overrides the service port address URL defined in the WSDL.
             if model.wsdl_endpoint_location:
                 soap_client.overridden_service = soap_client.create_service(
                     # picks the first binding in the WSDL as the default
                     soap_client.wsdl.bindings.keys()[0],
                     model.wsdl_endpoint_location)
             else:
                 soap_client.overridden_service = soap_client.service
             return soap_client
         except ConnectionError as exc:
             raise APIError('i-Parapheur error: %s' % exc)
     def format_type(t):
         return {'id': unicode(t), 'text': unicode(t)}
-...
         def get_verbose_name(cls):
             return cls._meta.verbose_name
         def call(self, service_name, *args, **kwargs):
             client = get_client(self)
         def get_client(self, strict=True):
             try:
                 soap_client = self.soap_client(strict=strict)
                 # overrides the service port address URL defined in the WSDL.
                 if self.wsdl_endpoint_location:
                     soap_client.overridden_service = soap_client.create_service(
                         # picks the first binding in the WSDL as the default
                         soap_client.wsdl.bindings.keys()[0],
                         self.wsdl_endpoint_location)
                 else:
                     soap_client.overridden_service = soap_client.service
                 return soap_client
             except ConnectionError as exc:
                 raise APIError('i-Parapheur error: %s' % exc)
         def call(self, service_name, args=[], kwargs={}, strict=True):
             client = self.get_client(strict=strict)
             try:
                 result = getattr(client.overridden_service, service_name)(*args, **kwargs)
             except WebFault as exc:
-...
                 raise APIError('Transport Error: %s' % exc)
             except TypeError as exc:
                 raise APIError('Type Error: %s' % exc)
             except XMLSyntaxError as exc:
                 raise APIError('XMLSyntax Error: %s' % exc)
             return result
         @endpoint(perm='can_access')
-...
         @endpoint(perm='can_access')
         def ping(self, request):
             return {'data': self.call('echo', 'ping')}
             return {'data': self.call('echo', args=['ping'])}
         @endpoint()
         def wsdl(self, request):
-...
         @endpoint(perm='can_access')
         def subtypes(self, request, type=None):
             if type:
                 return {'data': [format_type(t) for t in self.call('GetListeSousTypes', type)]}
                 return {'data': [format_type(t) for t in self.call('GetListeSousTypes', args=[type])]}
             return {'data': [format_type(t) for t in self.call('GetListeSousTypes')]}
         @endpoint(perm='can_access')
         def files(self, request, status=None):
             if status:
                 return {'data': [format_file(f) for f in self.call('RechercherDossiers', Status=status)]}
                 return {'data': [format_file(f) for f in self.call(
                     'RechercherDossiers', kwargs={'Status': status})]}
             return {'data': [format_file(f) for f in self.call('RechercherDossiers')]}
         @endpoint(
-...
                 raise APIError('Invalid base64 string')
             content_type = post_data['file']['content_type']
             soap_client = get_client(self)
             soap_client = self.get_client()
             if post_data['visibility'] not in ['PUBLIC', 'SERVICE', 'CONFIDENTIEL']:
                 raise FileError('Unknown value for "visibility". Should be "PUBLIC", "SERVICE" or "CONFIDENTIEL"')
-...
         @endpoint(perm='can_access', name='get-file', pattern='(?P<file_id>[\w-]+)')
         def get_file(self, request, file_id, appendix=None):
             resp = self.call('GetDossier', file_id)
             resp = self.call('GetDossier', args=[file_id], strict=False)
             filename = None
             if not resp or not resp.MessageRetour:
-...
         @endpoint(perm='can_access', name='get-file-status', pattern='(?P<file_id>[\w-]+)')
         def get_file_status(self, request, file_id):
             resp = self.call('GetHistoDossier', file_id)
             resp = self.call('GetHistoDossier', args=[file_id])
             if not resp or not resp.MessageRetour:
                 raise FileError('unknown error, no response')
             if resp.MessageRetour.codeRetour == 'KO':

         response.status_code = 200
         return response
     @mock.patch('passerelle.contrib.iparapheur.models.get_client')
     @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'
-...
         mocked_post.return_value = 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'
         assert 'zeep.exceptions.XMLParseError' in resp.json['err_class']
         assert resp.json['err_desc'] == "Unexpected element u'{http://www.adullact.org/spring-ws/iparapheur/1.0}MessageRetour', expected u'{http://www.adullact.org/spring-ws/iparapheur/1.0}TypeTechnique'"
         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>"""
-...
         soap_response = """<nada>"""
         response._content = soap_response
         mocked_post.return_value = response
         #resp = app.get(url, status=500)
         resp = app.get(url)
         assert resp.json['err'] == 1
         #assert 'zeep.exceptions.TransportError' in resp.json['err_class']
         assert 'passerelle.utils.jsonresponse.APIError' in resp.json['err_class']
         assert 'Server returned HTTP status 200 (<nada>)' in resp.json['err_desc']
         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')
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0001-iparapheur-get_file-allow-missing-element-in-xsd-seq.patch