0001-toulouse_maelis-use-responses-to-check-SOAP-messages.patch
| tests/test_toulouse_maelis.py | ||
|---|---|---|
| 18 | 18 |
from unittest import mock |
| 19 | 19 | |
| 20 | 20 |
import pytest |
| 21 |
import responses |
|
| 21 | 22 |
from lxml import etree |
| 22 | 23 |
from requests.exceptions import ConnectionError |
| 23 | 24 | |
| 24 | 25 |
from passerelle.contrib.toulouse_maelis.models import Link, Referential, ToulouseMaelis |
| 25 | 26 |
from passerelle.utils.jsonresponse import APIError |
| 26 | 27 |
from passerelle.utils.soap import SOAPError |
| 27 |
from tests.utils import FakedResponse, generic_endpoint_url, setup_access_rights |
|
| 28 |
from tests.utils import FakedResponse, ResponsesSoap, generic_endpoint_url, setup_access_rights
|
|
| 28 | 29 | |
| 29 | 30 |
TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'toulouse_maelis') |
| 30 | 31 | |
| ... | ... | |
| 277 | 278 |
assert Referential.objects.count() == 230 |
| 278 | 279 | |
| 279 | 280 | |
| 280 |
@mock.patch('passerelle.utils.Request.get')
|
|
| 281 |
@mock.patch('passerelle.utils.Request.post')
|
|
| 282 |
def test_link(mocked_post, mocked_get, con, app): |
|
| 283 |
mocked_get.return_value = FAMILY_SERVICE_WSDL |
|
| 284 |
mocked_post.return_value = READ_FAMILY |
|
| 281 |
@pytest.fixture |
|
| 282 |
def family_service(): |
|
| 283 |
from zeep import Settings |
|
| 284 | ||
| 285 |
with ResponsesSoap( |
|
| 286 |
wsdl_url='https://demo-toulouse.sigec.fr/maelisws-toulouse/services/FamilyService?wsdl', |
|
| 287 |
wsdl_content=get_xml_file('FamilyService.wsdl'),
|
|
| 288 |
settings=Settings(strict=False, xsd_ignore_sequence_order=True), |
|
| 289 |
)() as mock: |
|
| 290 |
yield mock |
|
| 291 | ||
| 292 | ||
| 293 |
def test_link(family_service, con, app): |
|
| 294 |
family_service.add_soap_response('readFamily', get_xml_file('R_read_family.xml'))
|
|
| 285 | 295 |
url = get_endpoint('link')
|
| 286 | 296 |
assert Link.objects.count() == 0 |
| 287 | 297 | |
| tests/utils.py | ||
|---|---|---|
| 1 |
import contextlib |
|
| 2 |
import io |
|
| 1 | 3 |
import json |
| 2 | 4 |
from unittest import mock |
| 3 | 5 |
from urllib import parse as urlparse |
| 4 | 6 | |
| 5 | 7 |
import httmock |
| 8 |
import lxml.etree as ET |
|
| 9 |
import responses |
|
| 10 |
import zeep |
|
| 11 |
import zeep.wsdl |
|
| 6 | 12 |
from django.contrib.contenttypes.models import ContentType |
| 7 | 13 |
from django.urls import reverse |
| 8 | 14 | |
| ... | ... | |
| 65 | 71 |
) |
| 66 | 72 |
assert url == expected_url, 'endpoint URL has changed' |
| 67 | 73 |
return app.get(url, **kwargs) |
| 74 | ||
| 75 | ||
| 76 |
class ResponsesSoap: |
|
| 77 |
def __init__(self, wsdl_url, wsdl_content, settings=None): |
|
| 78 |
self.wsdl_url = wsdl_url |
|
| 79 |
self.wsdl_content = wsdl_content |
|
| 80 |
self.wsdl = zeep.wsdl.Document(io.BytesIO(wsdl_content), None, settings=settings) |
|
| 81 |
self.soap_responses = [] |
|
| 82 |
assert ( |
|
| 83 |
len(self.wsdl.services.values()) == 1 |
|
| 84 |
), f'more or less than one service: {len(self.wsdl.bindings.values())}'
|
|
| 85 |
self.service = list(self.wsdl.services.values())[0] |
|
| 86 |
assert ( |
|
| 87 |
len(self.service.ports.values()) == 1 |
|
| 88 |
), f'more or less than one port: {len(self.service.ports.values())}'
|
|
| 89 |
self.port = list(self.service.ports.values())[0] |
|
| 90 |
self.binding = self.port.binding |
|
| 91 |
self.address = self.port.binding_options['address'] |
|
| 92 | ||
| 93 |
def soap_matcher(self, operation_name): |
|
| 94 |
operation = self.binding.get(operation_name) |
|
| 95 |
input_element_qname = operation.input.body.qname |
|
| 96 | ||
| 97 |
def matcher(prepared_request): |
|
| 98 |
doc = ET.parse(io.BytesIO(prepared_request.body)) |
|
| 99 |
if doc.find(f'.//{str(input_element_qname)}') is not None:
|
|
| 100 |
return True, f'Element "{str(input_element_qname)}" found'
|
|
| 101 |
return False, None |
|
| 102 | ||
| 103 |
return matcher |
|
| 104 | ||
| 105 |
def add_soap_response(self, mock, operation_name, response_content): |
|
| 106 |
operation = self.binding.get(operation_name) |
|
| 107 |
doc = ET.parse(io.BytesIO(response_content)) |
|
| 108 |
try: |
|
| 109 |
operation.output.deserialize(doc.getroot()) |
|
| 110 |
except Exception as e: |
|
| 111 |
raise AssertionError(f'response_content did not match operation "{operation_name}" schema') from e
|
|
| 112 |
mock.add( |
|
| 113 |
responses.POST, |
|
| 114 |
self.address, |
|
| 115 |
body=response_content, |
|
| 116 |
status=200, |
|
| 117 |
match=(self.soap_matcher(operation_name),), |
|
| 118 |
) |
|
| 119 | ||
| 120 |
@contextlib.contextmanager |
|
| 121 |
def __call__(self): |
|
| 122 |
with responses.RequestsMock() as mock: |
|
| 123 |
mock.add(responses.GET, self.wsdl_url, body=self.wsdl_content, status=200) |
|
| 124 |
mock.add_soap_response = lambda operation, response_content: self.add_soap_response( |
|
| 125 |
mock, operation, response_content |
|
| 126 |
) |
|
| 127 |
yield mock |
|
| 68 |
- |
|