Projet

Général

Profil

0001-tests-improve-ResponsesSoap-72638.patch

Benjamin Dauvergne, 21 décembre 2022 13:03

Télécharger (9,33 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
* allow nesting ResponsesSoap context manager by using the same
  RequestsMock context manager.

 tests/test_toulouse_maelis.py | 52 ++++++++++++++---------------------
 tests/utils.py                | 52 +++++++++++++++++------------------
 2 files changed, 46 insertions(+), 58 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)

  		




  102
  
  
    
            family_service.add_soap_response(

  		




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

  		




  104
  
  
    
            )

  		




  105
  
  
    
            family_service.add_soap_response(

  		




  106
  
  
    
                mock, 'readChildIndicatorList', get_xml_file('R_read_child_indicator_list.xml')

  		




  107
  
  
    
            )

  		




  108
  
  
    
            family_service.add_soap_response(

  		




  109
  
  
    
                mock, 'readCivilityList', get_xml_file('R_read_civility_list.xml')

  		




  110
  
  
    
            )

  		




  111
  
  
    
            family_service.add_soap_response(mock, 'readCountryList', get_xml_file('R_read_country_list.xml'))

  		




  112
  
  
    
            family_service.add_soap_response(mock, 'readCSPList', get_xml_file('R_read_csp_list.xml'))

  		




  113
  
  
    
            family_service.add_soap_response(

  		




  114
  
  
    
                mock, 'readDietCodeList', get_xml_file('R_read_dietcode_list.xml')

  		




  115
  
  
    
            )

  		




  116
  
  
    
            family_service.add_soap_response(mock, 'readOrganList', get_xml_file('R_read_organ_list.xml'))

  		




  117
  
  
    
            family_service.add_soap_response(mock, 'readPAIList', get_xml_file('R_read_pai_list.xml'))

  		




  118
  
  
    
            family_service.add_soap_response(mock, 'readQualityList', get_xml_file('R_read_quality_list.xml'))

  		




  119
  
  
    
            family_service.add_soap_response(

  		




  120
  
  
    
                mock, 'readQuotientList', get_xml_file('R_read_quotient_list.xml')

  		




  121
  
  
    
            )

  		




  
  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:

  		




  
  100
  
    
            family_service.add_soap_response('readCategoryList', get_xml_file('R_read_category_list.xml'))

  		




  122
  101
  
    
            family_service.add_soap_response(

  		




  123
  
  
    
                mock, 'readRLIndicatorList', get_xml_file('R_read_rl_indicator_list.xml')

  		




  
  102
  
    
                'readChildIndicatorList', get_xml_file('R_read_child_indicator_list.xml')

  		




  124
  103
  
    
            )

  		




  
  104
  
    
            family_service.add_soap_response('readCivilityList', get_xml_file('R_read_civility_list.xml'))

  		




  
  105
  
    
            family_service.add_soap_response('readCountryList', get_xml_file('R_read_country_list.xml'))

  		




  
  106
  
    
            family_service.add_soap_response('readCSPList', get_xml_file('R_read_csp_list.xml'))

  		




  
  107
  
    
            family_service.add_soap_response('readDietCodeList', get_xml_file('R_read_dietcode_list.xml'))

  		




  
  108
  
    
            family_service.add_soap_response('readOrganList', get_xml_file('R_read_organ_list.xml'))

  		




  
  109
  
    
            family_service.add_soap_response('readPAIList', get_xml_file('R_read_pai_list.xml'))

  		




  
  110
  
    
            family_service.add_soap_response('readQualityList', get_xml_file('R_read_quality_list.xml'))

  		




  
  111
  
    
            family_service.add_soap_response('readQuotientList', get_xml_file('R_read_quotient_list.xml'))

  		




  125
  112
  
    
            family_service.add_soap_response(

  		




  126
  
  
    
                mock, 'readSituationList', get_xml_file('R_read_situation_list.xml')

  		




  
  113
  
    
                'readRLIndicatorList', get_xml_file('R_read_rl_indicator_list.xml')

  		




  127
  114
  
    
            )

  		




  128
  
  
    
            family_service.add_soap_response(mock, 'readStreetList', get_xml_file('R_read_street_list.xml'))

  		




  129
  
  
    
            family_service.add_soap_response(mock, 'readVaccinList', get_xml_file('R_read_vaccin_list.xml'))

  		




  
  115
  
    
            family_service.add_soap_response('readSituationList', get_xml_file('R_read_situation_list.xml'))

  		




  
  116
  
    
            family_service.add_soap_response('readStreetList', get_xml_file('R_read_street_list.xml'))

  		




  
  117
  
    
            family_service.add_soap_response('readVaccinList', get_xml_file('R_read_vaccin_list.xml'))

  		




  130
  118
  
    
            con.daily()

  		




  131
  119
  
    

  		




  132
  120
  
    
    # reset change in zeep private interface to bypass clear_cache fixture

  		












  

    
      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, requests_mock=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']

  		




  
  94
  
    
        self.requests_mock = requests_mock or responses.RequestsMock()

  		




  
  95
  
    
        self.soap_requests = []

  		




  92
  96
  
    

  		




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

  		




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

  		




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

  		




  95
  99
  
    
        input_element_qname = operation.input.body.qname

  		




  96
  100
  
    

  		




  97
  101
  
    
        def matcher(prepared_request):

  		




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

  		




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

  		




  99
  104
  
    
            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)

  		




  
  105
  
    
                if request_check:

  		




  
  106
  
    
                    request_check(request)

  		




  
  107
  
    
                self.soap_requests.append(request)

  		




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

  		




  106
  109
  
    
            return False, None

  		




  107
  110
  
    

  		




  108
  111
  
    
        return matcher

  		




  109
  112
  
    

  		




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

  		




  
  113
  
    
    def add_soap_response(self, operation_name, response_content, status=200, request_check=None):

  		




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

  		




  112
  
  
    
        if not isinstance(response_content, Exception):

  		




  
  115
  
    
        if isinstance(response_content, dict):

  		




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

  		




  
  117
  
    
            body = ET.tostring(serialized_message.content)

  		




  
  118
  
    
        elif not isinstance(response_content, Exception):

  		




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

  		




  114
  120
  
    
            try:

  		




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

  		




  ......




  117
  123
  
    
                raise AssertionError(

  		




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

  		




  119
  125
  
    
                ) from e

  		




  120
  
  
    
        mock.add(

  		




  
  126
  
    
            body = response_content

  		




  
  127
  
    
        else:

  		




  
  128
  
    
            body = response_content

  		




  
  129
  
    
        self.requests_mock.add(

  		




  121
  130
  
    
            responses.POST,

  		




  122
  131
  
    
            self.address,

  		




  123
  
  
    
            body=response_content,

  		




  
  132
  
    
            body=body,

  		




  124
  133
  
    
            status=status,

  		




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

  		




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

  		




  126
  135
  
    
        )

  		




  127
  136
  
    

  		




  128
  137
  
    
    @contextlib.contextmanager

  		




  129
  138
  
    
    def __call__(self):

  		




  130
  
  
    
        with responses.RequestsMock() as mock:

  		




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

  		




  132
  
  
    
            mock.add_soap_response = (

  		




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

  		




  134
  
  
    
                    mock,

  		




  135
  
  
    
                    operation,

  		




  136
  
  
    
                    response_content,

  		




  137
  
  
    
                    status=status,

  		




  138
  
  
    
                    request_check=request_check,

  		




  139
  
  
    
                )

  		




  140
  
  
    
            )

  		




  141
  
  
    
            yield mock

  		




  
  139
  
    
        with self.requests_mock:

  		




  
  140
  
    
            self.requests_mock.add(responses.GET, self.wsdl_url, body=self.wsdl_content, status=200)

  		




  
  141
  
    
            yield self

  		




  142
  
  
    
-