entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 14 Releases Activity

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.
This commit is contained in:
Benjamin Dauvergne 2022-12-20 14:59:42 +01:00
parent db49dab374
commit c8155b70c4
2 changed files with 46 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
tests/test_toulouse_maelis.py
View File

@ -18,7 +18,6 @@ import os
from unittest import mock
import pytest
import responses
from requests.exceptions import ConnectionError
from zeep import Settings
@ -92,41 +91,30 @@ def django_db_setup(django_db_setup, django_db_blocker):
)[0]
)
with responses.RequestsMock() as mock:
family_service = ResponsesSoap(
wsdl_url='https://example.org/FamilyService?wsdl',
wsdl_content=get_xml_file('FamilyService.wsdl'),
settings=Settings(strict=False, xsd_ignore_sequence_order=True),
)
mock.add(responses.GET, family_service.wsdl_url, body=family_service.wsdl_content, status=200)
family_service = ResponsesSoap(
wsdl_url='https://example.org/FamilyService?wsdl',
wsdl_content=get_xml_file('FamilyService.wsdl'),
settings=Settings(strict=False, xsd_ignore_sequence_order=True),
)
with family_service():
family_service.add_soap_response('readCategoryList', get_xml_file('R_read_category_list.xml'))
family_service.add_soap_response(
mock, 'readCategoryList', get_xml_file('R_read_category_list.xml')
'readChildIndicatorList', get_xml_file('R_read_child_indicator_list.xml')
)
family_service.add_soap_response('readCivilityList', get_xml_file('R_read_civility_list.xml'))
family_service.add_soap_response('readCountryList', get_xml_file('R_read_country_list.xml'))
family_service.add_soap_response('readCSPList', get_xml_file('R_read_csp_list.xml'))
family_service.add_soap_response('readDietCodeList', get_xml_file('R_read_dietcode_list.xml'))
family_service.add_soap_response('readOrganList', get_xml_file('R_read_organ_list.xml'))
family_service.add_soap_response('readPAIList', get_xml_file('R_read_pai_list.xml'))
family_service.add_soap_response('readQualityList', get_xml_file('R_read_quality_list.xml'))
family_service.add_soap_response('readQuotientList', get_xml_file('R_read_quotient_list.xml'))
family_service.add_soap_response(
mock, 'readChildIndicatorList', get_xml_file('R_read_child_indicator_list.xml')
'readRLIndicatorList', get_xml_file('R_read_rl_indicator_list.xml')
)
family_service.add_soap_response(
mock, 'readCivilityList', get_xml_file('R_read_civility_list.xml')
)
family_service.add_soap_response(mock, 'readCountryList', get_xml_file('R_read_country_list.xml'))
family_service.add_soap_response(mock, 'readCSPList', get_xml_file('R_read_csp_list.xml'))
family_service.add_soap_response(
mock, 'readDietCodeList', get_xml_file('R_read_dietcode_list.xml')
)
family_service.add_soap_response(mock, 'readOrganList', get_xml_file('R_read_organ_list.xml'))
family_service.add_soap_response(mock, 'readPAIList', get_xml_file('R_read_pai_list.xml'))
family_service.add_soap_response(mock, 'readQualityList', get_xml_file('R_read_quality_list.xml'))
family_service.add_soap_response(
mock, 'readQuotientList', get_xml_file('R_read_quotient_list.xml')
)
family_service.add_soap_response(
mock, 'readRLIndicatorList', get_xml_file('R_read_rl_indicator_list.xml')
)
family_service.add_soap_response(
mock, 'readSituationList', get_xml_file('R_read_situation_list.xml')
)
family_service.add_soap_response(mock, 'readStreetList', get_xml_file('R_read_street_list.xml'))
family_service.add_soap_response(mock, 'readVaccinList', get_xml_file('R_read_vaccin_list.xml'))
family_service.add_soap_response('readSituationList', get_xml_file('R_read_situation_list.xml'))
family_service.add_soap_response('readStreetList', get_xml_file('R_read_street_list.xml'))
family_service.add_soap_response('readVaccinList', get_xml_file('R_read_vaccin_list.xml'))
con.daily()
# reset change in zeep private interface to bypass clear_cache fixture

52
tests/utils.py
View File

@ -74,9 +74,11 @@ def endpoint_get(expected_url, app, resource, endpoint, **kwargs):
class ResponsesSoap:
def __init__(self, wsdl_url, wsdl_content, settings=None):
def __init__(self, wsdl_url, wsdl_content, settings=None, address=None, requests_mock=None):
self.wsdl_url = wsdl_url
self.wsdl_content = wsdl_content
if isinstance(wsdl_content, str):
wsdl_content = wsdl_content.encode()
self.wsdl = zeep.wsdl.Document(io.BytesIO(wsdl_content), None, settings=settings)
self.soap_responses = []
assert (
@ -88,28 +90,32 @@ class ResponsesSoap:
), f'more or less than one port: {len(self.service.ports.values())}'
self.port = list(self.service.ports.values())[0]
self.binding = self.port.binding
self.address = self.port.binding_options['address']
self.address = address or self.port.binding_options['address']
self.requests_mock = requests_mock or responses.RequestsMock()
self.soap_requests = []
def soap_matcher(self, operation_name, request_check=None):
def soap_matcher(self, mock, operation_name, request_check=None):
operation = self.binding.get(operation_name)
input_element_qname = operation.input.body.qname
def matcher(prepared_request):
doc = ET.parse(io.BytesIO(prepared_request.body))
request = operation.input.deserialize(doc.getroot())
if doc.find(f'.//{str(input_element_qname)}') is not None:
try:
return True, f'Element "{str(input_element_qname)}" found'
finally:
if request_check:
request = operation.input.deserialize(doc.getroot())
request_check(request)
if request_check:
request_check(request)
self.soap_requests.append(request)
return True, f'Element "{str(input_element_qname)}" found'
return False, None
return matcher
def add_soap_response(self, mock, operation_name, response_content, status=200, request_check=None):
def add_soap_response(self, operation_name, response_content, status=200, request_check=None):
operation = self.binding.get(operation_name)
if not isinstance(response_content, Exception):
if isinstance(response_content, dict):
serialized_message = operation.output.serialize(**response_content)
body = ET.tostring(serialized_message.content)
elif not isinstance(response_content, Exception):
doc = ET.parse(io.BytesIO(response_content))
try:
operation.output.deserialize(doc.getroot())
@ -117,25 +123,19 @@ class ResponsesSoap:
raise AssertionError(
f'response_content did not match operation "{operation_name}" schema'
) from e
mock.add(
body = response_content
else:
body = response_content
self.requests_mock.add(
responses.POST,
self.address,
body=response_content,
body=body,
status=status,
match=(self.soap_matcher(operation_name, request_check),),
match=(self.soap_matcher(mock, operation_name, request_check),),
)
@contextlib.contextmanager
def __call__(self):
with responses.RequestsMock() as mock:
mock.add(responses.GET, self.wsdl_url, body=self.wsdl_content, status=200)
mock.add_soap_response = (
lambda operation, response_content, status=200, request_check=None: self.add_soap_response(
mock,
operation,
response_content,
status=status,
request_check=request_check,
)
)
yield mock
with self.requests_mock:
self.requests_mock.add(responses.GET, self.wsdl_url, body=self.wsdl_content, status=200)
yield self