From 37f0c0872652b7ed1c71b4fede8d3fe39c1d460b Mon Sep 17 00:00:00 2001 From: Benjamin Dauvergne Date: Tue, 20 Dec 2022 14:59:42 +0100 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(-) diff --git a/tests/test_toulouse_maelis.py b/tests/test_toulouse_maelis.py index 29872a6b..1012702f 100644 --- a/tests/test_toulouse_maelis.py +++ b/tests/test_toulouse_maelis.py @@ -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.add_soap_response( - mock, 'readCategoryList', get_xml_file('R_read_category_list.xml') - ) - family_service.add_soap_response( - mock, 'readChildIndicatorList', get_xml_file('R_read_child_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 = 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() as mock: + family_service.add_soap_response('readCategoryList', get_xml_file('R_read_category_list.xml')) family_service.add_soap_response( - mock, 'readRLIndicatorList', get_xml_file('R_read_rl_indicator_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, 'readSituationList', get_xml_file('R_read_situation_list.xml') + 'readRLIndicatorList', get_xml_file('R_read_rl_indicator_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 diff --git a/tests/utils.py b/tests/utils.py index a6f32a9e..13d1da3c 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -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 -- 2.37.2