0001-wip.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, 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 | |
| ... | ... | |
| 206 | 207 |
} |
| 207 | 208 | |
| 208 | 209 | |
| 209 |
@mock.patch('passerelle.utils.Request.get')
|
|
| 210 |
@mock.patch('passerelle.utils.Request.post')
|
|
| 211 |
def test_link(mocked_post, mocked_get, con, app): |
|
| 212 |
mocked_get.return_value = FAMILY_SERVICE_WSDL |
|
| 213 |
mocked_post.return_value = READ_FAMILY |
|
| 210 |
@pytest.fixture |
|
| 211 |
def family_service(): |
|
| 212 |
from zeep import Settings |
|
| 213 | ||
| 214 |
with ResponsesSoap( |
|
| 215 |
wsdl_url='https://demo-toulouse.sigec.fr/maelisws-toulouse/services/FamilyService?wsdl', |
|
| 216 |
wsdl_content=get_xml_file('FamilyService.wsdl'),
|
|
| 217 |
settings=Settings(strict=False, xsd_ignore_sequence_order=True), |
|
| 218 |
)() as mock: |
|
| 219 |
yield mock |
|
| 220 | ||
| 221 | ||
| 222 |
def test_link(family_service, con, app): |
|
| 223 |
family_service.add_soap_response('readFamily', get_xml_file('R_read_family.xml'))
|
|
| 214 | 224 |
url = get_endpoint('link')
|
| 215 | 225 |
assert Link.objects.count() == 0 |
| 216 | 226 | |
| 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 | |
| ... | ... | |
| 63 | 69 |
) |
| 64 | 70 |
assert url == expected_url, 'endpoint URL has changed' |
| 65 | 71 |
return app.get(url, **kwargs) |
| 72 | ||
| 73 | ||
| 74 |
class ResponsesSoap: |
|
| 75 |
def __init__(self, wsdl_url, wsdl_content, settings=None): |
|
| 76 |
self.wsdl_url = wsdl_url |
|
| 77 |
self.wsdl_content = wsdl_content |
|
| 78 |
self.wsdl = zeep.wsdl.Document(io.BytesIO(wsdl_content), None, settings=settings) |
|
| 79 |
self.soap_responses = [] |
|
| 80 |
assert ( |
|
| 81 |
len(self.wsdl.services.values()) == 1 |
|
| 82 |
), f'more or less than one service: {len(self.wsdl.bindings.values())}'
|
|
| 83 |
self.service = list(self.wsdl.services.values())[0] |
|
| 84 |
assert ( |
|
| 85 |
len(self.service.ports.values()) == 1 |
|
| 86 |
), f'more or less than one port: {len(self.service.ports.values())}'
|
|
| 87 |
self.port = list(self.service.ports.values())[0] |
|
| 88 |
self.binding = self.port.binding |
|
| 89 |
self.address = self.port.binding_options['address'] |
|
| 90 | ||
| 91 |
def soap_matcher(self, operation_name): |
|
| 92 |
operation = self.binding.get(operation_name) |
|
| 93 |
input_element_qname = operation.input.body.qname |
|
| 94 | ||
| 95 |
def matcher(prepared_request): |
|
| 96 |
doc = ET.parse(io.BytesIO(prepared_request.body)) |
|
| 97 |
if doc.find(f'.//{str(input_element_qname)}') is not None:
|
|
| 98 |
return True, f'Element "{str(input_element_qname)}" found'
|
|
| 99 |
return False, None |
|
| 100 | ||
| 101 |
return matcher |
|
| 102 | ||
| 103 |
def add_soap_response(self, mock, operation_name, response_content): |
|
| 104 |
operation = self.binding.get(operation_name) |
|
| 105 |
doc = ET.parse(io.BytesIO(response_content)) |
|
| 106 |
try: |
|
| 107 |
operation.output.deserialize(doc.getroot()) |
|
| 108 |
except Exception as e: |
|
| 109 |
raise AssertionError(f'response_content did not match operation "{operation_name}" schema') from e
|
|
| 110 |
mock.add( |
|
| 111 |
responses.POST, |
|
| 112 |
self.address, |
|
| 113 |
body=response_content, |
|
| 114 |
status=200, |
|
| 115 |
match=(self.soap_matcher(operation_name),), |
|
| 116 |
) |
|
| 117 | ||
| 118 |
@contextlib.contextmanager |
|
| 119 |
def __call__(self): |
|
| 120 |
with responses.RequestsMock() as mock: |
|
| 121 |
mock.add(responses.GET, self.wsdl_url, body=self.wsdl_content, status=200) |
|
| 122 |
mock.add_soap_response = lambda operation, response_content: self.add_soap_response( |
|
| 123 |
mock, operation, response_content |
|
| 124 |
) |
|
| 125 |
yield mock |
|
| 66 |
- |
|