Projet

Général

Profil

0001-tests-improve-ResponsesSoap-72638.patch

Benjamin Dauvergne, 20 décembre 2022 15:43

Télécharger (5,48 ko)

Voir les différences: 

Subject: [PATCH 1/2] tests: improve ResponsesSoap (#72638)

* add support for overriden service binding address
* encode wsdl_content when it is not bytes() but str()
* keep deserialized SOAP requests bodies in mock.soap_requests for
  inspection

 tests/test_toulouse_maelis.py | 14 ++++++--------
 tests/utils.py                | 32 ++++++++++++++++++++------------
 2 files changed, 26 insertions(+), 20 deletions(-)
tests/test_toulouse_maelis.py
18 18 
from unittest import mock
19 19 

  		




  20
  20
  
    
import pytest

  		




  21
  
  
    
import responses

  		




  22
  21
  
    
from requests.exceptions import ConnectionError

  		




  23
  22
  
    
from zeep import Settings

  		




  24
  23
  
    

  		




  ......




  92
  91
  
    
            )[0]

  		




  93
  92
  
    
        )

  		




  94
  93
  
    

  		




  95
  
  
    
        with responses.RequestsMock() as mock:

  		




  96
  
  
    
            family_service = ResponsesSoap(

  		




  97
  
  
    
                wsdl_url='https://example.org/FamilyService?wsdl',

  		




  98
  
  
    
                wsdl_content=get_xml_file('FamilyService.wsdl'),

  		




  99
  
  
    
                settings=Settings(strict=False, xsd_ignore_sequence_order=True),

  		




  100
  
  
    
            )

  		




  101
  
  
    
            mock.add(responses.GET, family_service.wsdl_url, body=family_service.wsdl_content, status=200)

  		




  
  94
  
    
        family_service = ResponsesSoap(

  		




  
  95
  
    
            wsdl_url='https://example.org/FamilyService?wsdl',

  		




  
  96
  
    
            wsdl_content=get_xml_file('FamilyService.wsdl'),

  		




  
  97
  
    
            settings=Settings(strict=False, xsd_ignore_sequence_order=True),

  		




  
  98
  
    
        )

  		




  
  99
  
    
        with family_service() as mock:

  		




  102
  100
  
    
            family_service.add_soap_response(

  		




  103
  101
  
    
                mock, 'readCategoryList', get_xml_file('R_read_category_list.xml')

  		




  104
  102
  
    
            )

  		












  

    
      tests/utils.py
    
  





  74
  74
  
    

  		




  75
  75
  
    

  		




  76
  76
  
    
class ResponsesSoap:

  		




  77
  
  
    
    def __init__(self, wsdl_url, wsdl_content, settings=None):

  		




  
  77
  
    
    def __init__(self, wsdl_url, wsdl_content, settings=None, address=None):

  		




  78
  78
  
    
        self.wsdl_url = wsdl_url

  		




  79
  79
  
    
        self.wsdl_content = wsdl_content

  		




  
  80
  
    
        if isinstance(wsdl_content, str):

  		




  
  81
  
    
            wsdl_content = wsdl_content.encode()

  		




  80
  82
  
    
        self.wsdl = zeep.wsdl.Document(io.BytesIO(wsdl_content), None, settings=settings)

  		




  81
  83
  
    
        self.soap_responses = []

  		




  82
  84
  
    
        assert (

  		




  ......




  88
  90
  
    
        ), f'more or less than one port: {len(self.service.ports.values())}'

  		




  89
  91
  
    
        self.port = list(self.service.ports.values())[0]

  		




  90
  92
  
    
        self.binding = self.port.binding

  		




  91
  
  
    
        self.address = self.port.binding_options['address']

  		




  
  93
  
    
        self.address = address or self.port.binding_options['address']

  		




  92
  94
  
    

  		




  93
  
  
    
    def soap_matcher(self, operation_name, request_check=None):

  		




  
  95
  
    
    def soap_matcher(self, mock, operation_name, request_check=None):

  		




  94
  96
  
    
        operation = self.binding.get(operation_name)

  		




  95
  97
  
    
        input_element_qname = operation.input.body.qname

  		




  96
  98
  
    

  		




  97
  99
  
    
        def matcher(prepared_request):

  		




  98
  100
  
    
            doc = ET.parse(io.BytesIO(prepared_request.body))

  		




  
  101
  
    
            request = operation.input.deserialize(doc.getroot())

  		




  99
  102
  
    
            if doc.find(f'.//{str(input_element_qname)}') is not None:

  		




  100
  
  
    
                try:

  		




  101
  
  
    
                    return True, f'Element "{str(input_element_qname)}" found'

  		




  102
  
  
    
                finally:

  		




  103
  
  
    
                    if request_check:

  		




  104
  
  
    
                        request = operation.input.deserialize(doc.getroot())

  		




  105
  
  
    
                        request_check(request)

  		




  
  103
  
    
                if request_check:

  		




  
  104
  
    
                    request_check(request)

  		




  
  105
  
    
                mock.soap_requests.append(request)

  		




  
  106
  
    
                return True, f'Element "{str(input_element_qname)}" found'

  		




  106
  107
  
    
            return False, None

  		




  107
  108
  
    

  		




  108
  109
  
    
        return matcher

  		




  109
  110
  
    

  		




  110
  111
  
    
    def add_soap_response(self, mock, operation_name, response_content, status=200, request_check=None):

  		




  111
  112
  
    
        operation = self.binding.get(operation_name)

  		




  112
  
  
    
        if not isinstance(response_content, Exception):

  		




  
  113
  
    
        if isinstance(response_content, dict):

  		




  
  114
  
    
            serialized_message = operation.output.serialize(**response_content)

  		




  
  115
  
    
            body = ET.tostring(serialized_message.content)

  		




  
  116
  
    
        elif not isinstance(response_content, Exception):

  		




  113
  117
  
    
            doc = ET.parse(io.BytesIO(response_content))

  		




  114
  118
  
    
            try:

  		




  115
  119
  
    
                operation.output.deserialize(doc.getroot())

  		




  ......




  117
  121
  
    
                raise AssertionError(

  		




  118
  122
  
    
                    f'response_content did not match operation "{operation_name}" schema'

  		




  119
  123
  
    
                ) from e

  		




  
  124
  
    
            body = response_content

  		




  
  125
  
    
        else:

  		




  
  126
  
    
            body = response_content

  		




  120
  127
  
    
        mock.add(

  		




  121
  128
  
    
            responses.POST,

  		




  122
  129
  
    
            self.address,

  		




  123
  
  
    
            body=response_content,

  		




  
  130
  
    
            body=body,

  		




  124
  131
  
    
            status=status,

  		




  125
  
  
    
            match=(self.soap_matcher(operation_name, request_check),),

  		




  
  132
  
    
            match=(self.soap_matcher(mock, operation_name, request_check),),

  		




  126
  133
  
    
        )

  		




  127
  134
  
    

  		




  128
  135
  
    
    @contextlib.contextmanager

  		




  129
  136
  
    
    def __call__(self):

  		




  130
  137
  
    
        with responses.RequestsMock() as mock:

  		




  
  138
  
    
            mock.soap_requests = []

  		




  131
  139
  
    
            mock.add(responses.GET, self.wsdl_url, body=self.wsdl_content, status=200)

  		




  132
  140
  
    
            mock.add_soap_response = (

  		




  133
  141
  
    
                lambda operation, response_content, status=200, request_check=None: self.add_soap_response(

  		




  134
  
  
    
-