From a187850b6d1d5ed2f8827ae38062eb7dc62b166f Mon Sep 17 00:00:00 2001
From: Benjamin Dauvergne <bdauvergne@entrouvert.com>
Date: Thu, 8 Dec 2022 22:22:31 +0100
Subject: [PATCH 1/3] toulouse_maelis: use responses to check SOAP messages
 (#72205)

---
 tests/test_toulouse_maelis.py | 22 +++++++++----
 tests/utils.py                | 60 +++++++++++++++++++++++++++++++++++
 2 files changed, 76 insertions(+), 6 deletions(-)

diff --git a/tests/test_toulouse_maelis.py b/tests/test_toulouse_maelis.py
index 1137efe0..a0b18358 100644
--- a/tests/test_toulouse_maelis.py
+++ b/tests/test_toulouse_maelis.py
@@ -18,13 +18,14 @@ import os
 from unittest import mock
 
 import pytest
+import responses
 from lxml import etree
 from requests.exceptions import ConnectionError
 
 from passerelle.contrib.toulouse_maelis.models import Link, Referential, ToulouseMaelis
 from passerelle.utils.jsonresponse import APIError
 from passerelle.utils.soap import SOAPError
-from tests.utils import FakedResponse, generic_endpoint_url, setup_access_rights
+from tests.utils import FakedResponse, ResponsesSoap, generic_endpoint_url, setup_access_rights
 
 TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'toulouse_maelis')
 
@@ -277,11 +278,20 @@ def test_cron(db):
     assert Referential.objects.count() == 230
 
 
-@mock.patch('passerelle.utils.Request.get')
-@mock.patch('passerelle.utils.Request.post')
-def test_link(mocked_post, mocked_get, con, app):
-    mocked_get.return_value = FAMILY_SERVICE_WSDL
-    mocked_post.return_value = READ_FAMILY
+@pytest.fixture
+def family_service():
+    from zeep import Settings
+
+    with ResponsesSoap(
+        wsdl_url='https://demo-toulouse.sigec.fr/maelisws-toulouse/services/FamilyService?wsdl',
+        wsdl_content=get_xml_file('FamilyService.wsdl'),
+        settings=Settings(strict=False, xsd_ignore_sequence_order=True),
+    )() as mock:
+        yield mock
+
+
+def test_link(family_service, con, app):
+    family_service.add_soap_response('readFamily', get_xml_file('R_read_family.xml'))
     url = get_endpoint('link')
     assert Link.objects.count() == 0
 
diff --git a/tests/utils.py b/tests/utils.py
index cd2065aa..26d7dc49 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -1,8 +1,14 @@
+import contextlib
+import io
 import json
 from unittest import mock
 from urllib import parse as urlparse
 
 import httmock
+import lxml.etree as ET
+import responses
+import zeep
+import zeep.wsdl
 from django.contrib.contenttypes.models import ContentType
 from django.urls import reverse
 
@@ -65,3 +71,57 @@ def endpoint_get(expected_url, app, resource, endpoint, **kwargs):
     )
     assert url == expected_url, 'endpoint URL has changed'
     return app.get(url, **kwargs)
+
+
+class ResponsesSoap:
+    def __init__(self, wsdl_url, wsdl_content, settings=None):
+        self.wsdl_url = wsdl_url
+        self.wsdl_content = wsdl_content
+        self.wsdl = zeep.wsdl.Document(io.BytesIO(wsdl_content), None, settings=settings)
+        self.soap_responses = []
+        assert (
+            len(self.wsdl.services.values()) == 1
+        ), f'more or less than one service: {len(self.wsdl.bindings.values())}'
+        self.service = list(self.wsdl.services.values())[0]
+        assert (
+            len(self.service.ports.values()) == 1
+        ), 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']
+
+    def soap_matcher(self, operation_name):
+        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))
+            if doc.find(f'.//{str(input_element_qname)}') is not None:
+                return True, f'Element "{str(input_element_qname)}" found'
+            return False, None
+
+        return matcher
+
+    def add_soap_response(self, mock, operation_name, response_content):
+        operation = self.binding.get(operation_name)
+        doc = ET.parse(io.BytesIO(response_content))
+        try:
+            operation.output.deserialize(doc.getroot())
+        except Exception as e:
+            raise AssertionError(f'response_content did not match operation "{operation_name}" schema') from e
+        mock.add(
+            responses.POST,
+            self.address,
+            body=response_content,
+            status=200,
+            match=(self.soap_matcher(operation_name),),
+        )
+
+    @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: self.add_soap_response(
+                mock, operation, response_content
+            )
+            yield mock
-- 
2.37.2