0005-wip-on-soaperror-set-availability-of-connector-to-do.patch

Benjamin Dauvergne, 23 mars 2022 11:14

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

Subject: [PATCH 5/7] wip: on soaperror set availability of connector to down
 when retrieving the WSDL schema

 passerelle/apps/soap/models.py | 21 +++++++++++++++++++--
 passerelle/base/models.py      |  4 ++--
 tests/test_soap.py             | 10 ++++++++++
 3 files changed, 31 insertions(+), 4 deletions(-)

     import zeep.helpers
     import zeep.xsd
     from django.db import models
     from django.forms import ValidationError
     from django.utils.functional import cached_property
     from django.utils.translation import ugettext_lazy as _
     from passerelle.base.models import BaseResource, HTTPResource
     from passerelle.utils.api import endpoint
     from passerelle.utils.conversion import exception_to_text
     from passerelle.utils.json import unflatten
-...
         class Meta:
             verbose_name = _('SOAP connector')
         def clean(self):
             try:
                 self.operations_and_schemas
             except Exception as e:
                 raise ValidationError(e)
         @classmethod
         def get_manager_form_class(cls, **kwargs):
             form_class = super().get_manager_form_class(**kwargs)
-...
         def get_endpoints_infos(self):
             endpoints = super().get_endpoints_infos()
             for name, input_schema, output_schema in self.operations_and_schemas:
             try:
                 operations_and_schemas = self.operations_and_schemas
             except Exception as e:
                 self.set_availability_status('down', message=exception_to_text(e)[:500])
                 return endpoints
             for name, input_schema, output_schema in operations_and_schemas:
                 kwargs = dict(
                     name='method',
                     pattern=f'{name}/',
-...
         @property
         def operations_and_schemas(self):
             operations = self.client.service._binding._operations
             operations_and_schemas = []
             for name in operations:
                 operation = operations[name]
                 input_type = operation.input.body.type
                 output_type = operation.output.body.type
                 input_schema = self.type2schema(input_type, keep_root=True)
                 output_schema = self.type2schema(output_type, compress=True)
                 yield name, input_schema, output_schema
                 operations_and_schemas.append((name, input_schema, output_schema))
             return operations_and_schemas
         def type2schema(self, xsd_type, keep_root=False, compress=False):
             # simplify schema: when a type contains a unique element, it will try

         def get_endpoints_infos(self):
             endpoints = []
             for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
             for name, method in inspect.getmembers(type(self), predicate=inspect.isfunction):
                 if hasattr(method, 'endpoint_info'):
                     method.endpoint_info.object = self
                     endpoint_name = method.endpoint_info.name
-...
             except Exception as e:
                 from passerelle.utils.conversion import exception_to_text
                 self.set_availability('down', message=exception_to_text(e)[:500])
                 self.set_availability_status('down', message=exception_to_text(e)[:500])
             else:
                 self.set_availability_status('up')

+        }
     class BrokenSOAP12(SOAP12):
         WSDL_CONTENT = SOAP12.WSDL_CONTENT[-100:]  # truncate the WSDL to break it
     @pytest.fixture(params=[SOAP11, SOAP12])
     def soap(request):
         p = request.param()
-...
             response = app.get(f'/soap/{connector.slug}/')
             assert 'Method sayHello' in response
         @pytest.mark.parametrize('soap', [BrokenSOAP12], indirect=True)
         def test_homepage_broken_wsdl(self, app, connector, soap):
             response = app.get(f'/soap/{connector.slug}/')
             response = app.get(f'/soap/{connector.slug}/')
             assert response.pyquery('.down')
     @pytest.fixture
     def connector(db, soap):
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0005-wip-on-soaperror-set-availability-of-connector-to-do.patch