From f7f757aa2a77f909896199de83b0efb45e972c74 Mon Sep 17 00:00:00 2001
From: Benjamin Dauvergne <bdauvergne@entrouvert.com>
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
---
 tests/test_toulouse_maelis.py | 13 ++++++-------
 tests/utils.py                | 32 ++++++++++++++++++++------------
 2 files changed, 26 insertions(+), 19 deletions(-)

diff --git a/tests/test_toulouse_maelis.py b/tests/test_toulouse_maelis.py
index 29872a6b..fedca952 100644
--- a/tests/test_toulouse_maelis.py
+++ b/tests/test_toulouse_maelis.py
@@ -92,13 +92,12 @@ 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() as mock:
             family_service.add_soap_response(
                 mock, 'readCategoryList', get_xml_file('R_read_category_list.xml')
             )
diff --git a/tests/utils.py b/tests/utils.py
index a6f32a9e..7d69bf6f 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):
         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,30 @@ 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']
 
-    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)
+                mock.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):
         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,17 +121,21 @@ class ResponsesSoap:
                 raise AssertionError(
                     f'response_content did not match operation "{operation_name}" schema'
                 ) from e
+            body = response_content
+        else:
+            body = response_content
         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.soap_requests = []
             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(
-- 
2.37.2