142 lines
5.0 KiB
Python
142 lines
5.0 KiB
Python
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
|
|
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
|
|
|
|
def generic_endpoint_url(connector, endpoint, slug='test'):
|
|
return reverse('generic-endpoint', kwargs={'connector': connector, 'slug': slug, 'endpoint': endpoint})
|
|
|
|
|
|
def setup_access_rights(obj):
|
|
api, _ = ApiUser.objects.get_or_create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(obj)
|
|
AccessRight.objects.update_or_create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=obj.pk
|
|
)
|
|
return obj
|
|
|
|
|
|
class FakedResponse(mock.Mock):
|
|
headers = {}
|
|
|
|
def json(self):
|
|
return json.loads(self.content)
|
|
|
|
|
|
def mock_url(url=None, response='', status_code=200, headers=None, reason=None, exception=None, qs=None):
|
|
urlmatch_kwargs = {}
|
|
if url:
|
|
parsed = urlparse.urlparse(url)
|
|
if parsed.netloc:
|
|
urlmatch_kwargs['netloc'] = parsed.netloc
|
|
if parsed.path:
|
|
urlmatch_kwargs['path'] = parsed.path
|
|
|
|
if not isinstance(response, str):
|
|
response = json.dumps(response)
|
|
|
|
@httmock.remember_called
|
|
@httmock.urlmatch(**urlmatch_kwargs)
|
|
def mocked(url, request):
|
|
if qs is not None:
|
|
qs.update(urlparse.parse_qsl(url.query))
|
|
if exception:
|
|
raise exception
|
|
return httmock.response(status_code, response, headers, reason, request=request)
|
|
|
|
return httmock.HTTMock(mocked)
|
|
|
|
|
|
def make_resource(model_class, **kwargs):
|
|
resource = model_class.objects.create(**kwargs)
|
|
setup_access_rights(resource)
|
|
return resource
|
|
|
|
|
|
def endpoint_get(expected_url, app, resource, endpoint, **kwargs):
|
|
url = generic_endpoint_url(
|
|
connector=resource.__class__.get_connector_slug(), endpoint=endpoint, slug=resource.slug
|
|
)
|
|
assert url == expected_url, 'endpoint URL has changed'
|
|
return app.get(url, **kwargs)
|
|
|
|
|
|
class ResponsesSoap:
|
|
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 (
|
|
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 = address or self.port.binding_options['address']
|
|
self.requests_mock = requests_mock or responses.RequestsMock()
|
|
self.soap_requests = []
|
|
|
|
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:
|
|
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, operation_name, response_content, status=200, request_check=None):
|
|
operation = self.binding.get(operation_name)
|
|
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())
|
|
except Exception as e:
|
|
raise AssertionError(
|
|
f'response_content did not match operation "{operation_name}" schema'
|
|
) from e
|
|
body = response_content
|
|
else:
|
|
body = response_content
|
|
return self.requests_mock.add(
|
|
responses.POST,
|
|
self.address,
|
|
body=body,
|
|
status=status,
|
|
match=(self.soap_matcher(mock, operation_name, request_check),),
|
|
)
|
|
|
|
@contextlib.contextmanager
|
|
def __call__(self):
|
|
with self.requests_mock:
|
|
self.requests_mock.add(responses.GET, self.wsdl_url, body=self.wsdl_content, status=200)
|
|
yield self
|