diff --git a/passerelle/apps/soap/__init__.py b/passerelle/apps/soap/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/passerelle/apps/soap/migrations/0001_initial.py b/passerelle/apps/soap/migrations/0001_initial.py
new file mode 100644
index 00000000..6285d7a7
--- /dev/null
+++ b/passerelle/apps/soap/migrations/0001_initial.py
@@ -0,0 +1,91 @@
+# Generated by Django 2.2.24 on 2022-01-19 16:37
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ initial = True
+
+ dependencies = [
+ ('base', '0029_auto_20210202_1627'),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name='SOAPConnector',
+ fields=[
+ (
+ 'id',
+ models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
+ ),
+ ('title', models.CharField(max_length=50, verbose_name='Title')),
+ ('slug', models.SlugField(unique=True, verbose_name='Identifier')),
+ ('description', models.TextField(verbose_name='Description')),
+ (
+ 'basic_auth_username',
+ models.CharField(
+ blank=True, max_length=128, verbose_name='Basic authentication username'
+ ),
+ ),
+ (
+ 'basic_auth_password',
+ models.CharField(
+ blank=True, max_length=128, verbose_name='Basic authentication password'
+ ),
+ ),
+ (
+ 'client_certificate',
+ models.FileField(
+ blank=True, null=True, upload_to='', verbose_name='TLS client certificate'
+ ),
+ ),
+ (
+ 'trusted_certificate_authorities',
+ models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'),
+ ),
+ (
+ 'verify_cert',
+ models.BooleanField(blank=True, default=True, verbose_name='TLS verify certificates'),
+ ),
+ (
+ 'http_proxy',
+ models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'),
+ ),
+ (
+ 'wsdl_url',
+ models.URLField(
+ help_text='URL of the WSDL file', max_length=400, verbose_name='WSDL URL'
+ ),
+ ),
+ (
+ 'zeep_strict',
+ models.BooleanField(default=True, verbose_name='Be strict with returned XML'),
+ ),
+ (
+ 'zeep_xsd_ignore_sequence_order',
+ models.BooleanField(default=False, verbose_name='Ignore sequence order'),
+ ),
+ (
+ 'zeep_wsse_username',
+ models.CharField(max_length=256, blank=True, default='', verbose_name='WSSE Username'),
+ ),
+ (
+ 'zeep_wsse_password',
+ models.CharField(max_length=256, blank=True, default='', verbose_name='WSSE Password'),
+ ),
+ (
+ 'users',
+ models.ManyToManyField(
+ blank=True,
+ related_name='_soapconnector_users_+',
+ related_query_name='+',
+ to='base.ApiUser',
+ ),
+ ),
+ ],
+ options={
+ 'verbose_name': 'SOAP connector',
+ },
+ ),
+ ]
diff --git a/passerelle/apps/soap/migrations/__init__.py b/passerelle/apps/soap/migrations/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/passerelle/apps/soap/models.py b/passerelle/apps/soap/models.py
new file mode 100644
index 00000000..1d0e3164
--- /dev/null
+++ b/passerelle/apps/soap/models.py
@@ -0,0 +1,224 @@
+# passerelle - uniform access to multiple data sources and services
+# Copyright (C) 2019 Entr'ouvert
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see .
+
+import collections
+
+import zeep
+import zeep.exceptions
+import zeep.helpers
+import zeep.xsd
+from django.db import models
+from django.forms import ValidationError
+from django.utils.functional import cached_property
+from django.utils.translation import ugettext_lazy as _
+from zeep.wsse.username import UsernameToken
+
+from passerelle.base.models import BaseResource, HTTPResource
+from passerelle.utils.api import endpoint
+from passerelle.utils.conversion import exception_to_text
+from passerelle.utils.json import unflatten
+from passerelle.utils.jsonresponse import APIError
+
+
+class SOAPConnector(BaseResource, HTTPResource):
+ wsdl_url = models.URLField(
+ max_length=400, verbose_name=_('WSDL URL'), help_text=_('URL of the WSDL file')
+ )
+ zeep_strict = models.BooleanField(default=False, verbose_name=_('Be strict with returned XML'))
+ zeep_xsd_ignore_sequence_order = models.BooleanField(
+ default=True, verbose_name=_('Ignore sequence order')
+ )
+ zeep_wsse_username = models.CharField(
+ max_length=256, blank=True, default='', verbose_name=_('WSSE Username')
+ )
+ zeep_wsse_password = models.CharField(
+ max_length=256, blank=True, default='', verbose_name=_('WSSE Password')
+ )
+ category = _('Business Process Connectors')
+
+ class Meta:
+ verbose_name = _('SOAP connector')
+
+ def clean(self):
+ try:
+ self.operations_and_schemas
+ except Exception as e:
+ raise ValidationError(e)
+
+ @classmethod
+ def get_manager_form_class(cls, **kwargs):
+ exclude = kwargs.get('exclude')
+ form_class = super().get_manager_form_class(**kwargs)
+ fields = list(form_class.base_fields.items())
+ if exclude and 'slug' in exclude:
+ form_class.base_fields = collections.OrderedDict(fields[:2] + fields[-5:] + fields[2:-5])
+ else:
+ form_class.base_fields = collections.OrderedDict(fields[:3] + fields[-5:] + fields[3:-5])
+ return form_class
+
+ @cached_property
+ def client(self):
+ kwargs = {}
+ if self.zeep_wsse_username:
+ kwargs['wsse'] = UsernameToken(self.zeep_wsse_username, self.zeep_wsse_password)
+ return self.soap_client(
+ wsdl_url=self.wsdl_url,
+ settings=zeep.Settings(
+ strict=self.zeep_strict, xsd_ignore_sequence_order=self.zeep_xsd_ignore_sequence_order
+ ),
+ **kwargs,
+ )
+
+ @endpoint(
+ methods=['post'],
+ perm='can_access',
+ name='method',
+ pattern=r'^(?P\w+)/$',
+ example_pattern='method_name/',
+ description_get=_('Call a SOAP method'),
+ description_post=_('Call a SOAP method'),
+ post_json_schema={'type': 'object'},
+ )
+ def method(self, request, method_name, post_data=None, **kwargs):
+ def jsonify(data):
+ if isinstance(data, (dict, collections.OrderedDict)):
+ # ignore _raw_elements, zeep put there nodes not maching the
+ # XSD when strict parsing is disabled.
+ return {
+ jsonify(k): jsonify(v)
+ for k, v in data.items()
+ if (self.zeep_strict or k != '_raw_elements')
+ }
+ elif isinstance(data, (list, tuple, collections.deque)):
+ return [jsonify(item) for item in data]
+ else:
+ return data
+
+ payload = {}
+ for k in request.GET:
+ if k == 'raise':
+ continue
+ value = request.GET.getlist(k)
+ if len(value) > 1:
+ payload[k] = value
+ else:
+ payload[k] = value[0]
+ payload.update(post_data or {})
+ payload = unflatten(payload)
+ try:
+ soap_response = getattr(self.client.service, method_name)(**payload)
+ except zeep.exceptions.Fault as e:
+ fault_details = {}
+ for attrib in ['actor', 'code', 'message', 'subcode']:
+ fault_details[attrib] = getattr(e, attrib, None)
+ raise APIError('soap:Fault', data=fault_details)
+ except zeep.exceptions.ValidationError as e:
+ e.status_code = 400
+ raise e
+ serialized = zeep.helpers.serialize_object(soap_response)
+ json_response = jsonify(serialized)
+ return {'err': 0, 'data': json_response}
+
+ method.endpoint_info.methods.append('get')
+
+ def get_endpoints_infos(self):
+ endpoints = super().get_endpoints_infos()
+
+ try:
+ operations_and_schemas = self.operations_and_schemas
+ except Exception as e:
+ self.set_availability_status('down', message=exception_to_text(e)[:500])
+ return endpoints
+
+ for name, input_schema, output_schema in operations_and_schemas:
+ kwargs = dict(
+ name='method',
+ pattern=f'{name}/',
+ example_pattern=f'{name}/',
+ description=f'Method {name}',
+ json_schema_response={
+ 'type': 'object',
+ 'properties': collections.OrderedDict(
+ [
+ ('err', {'type': 'integer'}),
+ ('data', output_schema),
+ ]
+ ),
+ },
+ )
+ if input_schema:
+ kwargs['post_json_schema'] = input_schema
+ endpoints.append(endpoint(**kwargs))
+ endpoints[-1].object = self
+ endpoints[-1].func = lambda request: None
+ if input_schema and input_schema.get('properties'):
+ endpoints[-1].http_method = 'post'
+ else:
+ endpoints[-1].http_method = 'get'
+ return endpoints
+
+ @property
+ def operations_and_schemas(self):
+ operations = self.client.service._binding._operations
+ operations_and_schemas = []
+ for name in operations:
+ operation = operations[name]
+ input_type = operation.input.body.type
+ output_type = operation.output.body.type
+ input_schema = self.type2schema(input_type, keep_root=True)
+ output_schema = self.type2schema(output_type, compress=True)
+ operations_and_schemas.append((name, input_schema, output_schema))
+ return operations_and_schemas
+
+ def type2schema(self, xsd_type, keep_root=False, compress=False):
+ # simplify schema: when a type contains a unique element, it will try
+ # to match any dict or list with it on input and will flatten the
+ # schema on output.
+ if (
+ isinstance(xsd_type, zeep.xsd.ComplexType)
+ and len(xsd_type.elements) == 1
+ and not keep_root
+ and compress
+ ):
+ if xsd_type.elements[0][1].max_occurs != 1:
+ schema = {
+ 'type': 'array',
+ 'items': self.type2schema(xsd_type.elements[0][1].type, compress=compress),
+ }
+ else:
+ schema = self.type2schema(xsd_type.elements[0][1].type, compress=compress)
+ elif isinstance(xsd_type, zeep.xsd.ComplexType):
+ properties = collections.OrderedDict()
+ schema = {
+ 'type': 'object',
+ 'properties': properties,
+ }
+ for key, element in xsd_type.elements:
+ if element.min_occurs > 0:
+ schema.setdefault('required', []).append(key)
+ element_schema = self.type2schema(element.type, compress=compress)
+ if element.max_occurs == 'unbounded' or element.max_occurs > 1:
+ element_schema = {'type': 'array', 'items': element_schema}
+ properties[key] = element_schema
+ if not properties:
+ schema = {'type': 'null'}
+ elif isinstance(xsd_type, zeep.xsd.BuiltinType):
+ schema = {'type': 'string'}
+ else:
+ schema = {}
+ if xsd_type.qname:
+ schema['description'] = str(xsd_type.qname).replace('{http://www.w3.org/2001/XMLSchema}', 'xsd:')
+ return schema
diff --git a/passerelle/apps/soap/templates/soap/soapconnector_detail.html b/passerelle/apps/soap/templates/soap/soapconnector_detail.html
new file mode 100644
index 00000000..67a705ec
--- /dev/null
+++ b/passerelle/apps/soap/templates/soap/soapconnector_detail.html
@@ -0,0 +1,2 @@
+{% extends "passerelle/manage/service_view.html" %}
+{% load i18n passerelle %}
diff --git a/passerelle/settings.py b/passerelle/settings.py
index 0152f331..b51442e5 100644
--- a/passerelle/settings.py
+++ b/passerelle/settings.py
@@ -162,6 +162,7 @@ INSTALLED_APPS = (
'passerelle.apps.plone_restapi',
'passerelle.apps.sector',
'passerelle.apps.sfr_dmc',
+ 'passerelle.apps.soap',
'passerelle.apps.solis',
'passerelle.apps.sp_fr',
'passerelle.apps.twilio',
diff --git a/tests/test_soap.py b/tests/test_soap.py
index 35fc9472..5634b264 100644
--- a/tests/test_soap.py
+++ b/tests/test_soap.py
@@ -13,97 +13,342 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
-import mock
+import base64
+import urllib.parse
+
import pytest
-import requests
-from django.utils.encoding import force_bytes
-from zeep import Settings
-from zeep.exceptions import TransportError, XMLParseError
-from zeep.plugins import Plugin
-from passerelle.utils.soap import SOAPClient
+from passerelle.apps.soap.models import SOAPConnector
-WSDL = 'tests/data/soap.wsdl'
+from . import utils
+
+pytestmark = pytest.mark.django_db
-class FooPlugin(Plugin):
- pass
+class SOAP11:
+ VERSION = '1.1'
+ ENDPOINT_URL = 'https://www.examples.com/SayHello/'
+ WSDL_CONTENT = '''\
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ WSDL File for HelloService
+
+
+
+
+'''
+ WSDL_URL = 'https://example.com/service.wsdl'
+ SOAP_RESPONSE = '''\
+
+
+
+
+ Hello
+ John!
+
+
+'''
+ INPUT_SCHEMA = {
+ 'properties': {
+ 'firstName': {
+ 'description': '{http://www.examples.com/wsdl/HelloService.wsdl}firstName',
+ 'properties': {
+ 'string': {'items': {'type': 'string', 'description': 'xsd:string'}, 'type': 'array'},
+ },
+ 'required': ['string'],
+ 'type': 'object',
+ },
+ 'lastName': {'type': 'string', 'description': 'xsd:string'},
+ },
+ 'required': ['firstName', 'lastName'],
+ 'type': 'object',
+ }
+ OUTPUT_SCHEMA = {
+ 'properties': {
+ 'greeting': {'type': 'string', 'description': 'xsd:string'},
+ 'who': {'type': 'string', 'description': 'xsd:string'},
+ },
+ 'required': ['greeting', 'who'],
+ 'type': 'object',
+ }
+ INPUT_DATA = {
+ 'firstName/string/0': 'John',
+ 'firstName/string/1': 'Bill',
+ 'lastName': 'Doe',
+ }
+ OUTPUT_DATA = {
+ 'greeting': 'Hello',
+ 'who': 'John!',
+ }
-class BarPlugin(Plugin):
- pass
+class SOAP12(SOAP11):
+ VERSION = '1.2'
+ ENDPOINT_URL = 'https://www.examples.com/SayHello/'
+ WSDL_CONTENT = f'''\
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+'''
+ SOAP_RESPONSE = '''\
+
+
+
+
+
+ Hello
+ John!
+
+
+'''
+ INPUT_SCHEMA = {
+ 'type': 'object',
+ 'properties': {
+ 'firstName': {'type': 'array', 'items': {'type': 'string', 'description': 'xsd:string'}},
+ 'lastName': {'type': 'string', 'description': 'xsd:string'},
+ },
+ 'required': ['firstName', 'lastName'],
+ 'description': '{urn:examples:helloservice}sayHello',
+ }
+ OUTPUT_SCHEMA = {
+ 'description': '{urn:examples:helloservice}sayHelloResponse',
+ 'properties': {
+ 'greeting': {'type': 'string', 'description': 'xsd:string'},
+ 'who': {
+ 'type': 'array',
+ 'items': {'type': 'string', 'description': 'xsd:string'},
+ },
+ },
+ 'required': ['greeting', 'who'],
+ 'type': 'object',
+ }
+ INPUT_DATA = {
+ 'firstName/0': 'John',
+ 'firstName/1': 'Bill',
+ 'lastName': 'Doe',
+ }
+ OUTPUT_DATA = {
+ 'greeting': 'Hello',
+ 'who': ['John!'],
+ }
-class SOAPResource:
- def __init__(self):
- self.requests = requests.Session()
- self.wsdl_url = WSDL
+class BrokenSOAP12(SOAP12):
+ WSDL_CONTENT = SOAP12.WSDL_CONTENT[-100:] # truncate the WSDL to break it
-def test_soap_client():
- soap_resource = SOAPResource()
- plugins = [FooPlugin, BarPlugin]
- client = SOAPClient(soap_resource, plugins=plugins)
- assert client.wsdl.location.endswith(WSDL)
- assert client.transport.session == soap_resource.requests
- assert client.transport.cache
- assert client.plugins == plugins
+@pytest.fixture(params=[SOAP11, SOAP12])
+def soap(request):
+ p = request.param()
+ with utils.mock_url(p.WSDL_URL, response=p.WSDL_CONTENT):
+ with utils.mock_url(p.ENDPOINT_URL, response=p.SOAP_RESPONSE) as mock:
+ p.endpoint_mock = mock
+ yield p
-@mock.patch('requests.sessions.Session.post')
-def test_disable_strict_mode(mocked_post):
- response = requests.Response()
- response.status_code = 200
- response._content = force_bytes(
- '''
-
-
-
- 4.20
-
-
-'''
+class TestManage:
+ @pytest.fixture
+ def app(self, app, admin_user):
+ from .test_manager import login
+
+ login(app)
+ return app
+
+ def test_homepage(self, app, connector, soap):
+ response = app.get(f'/soap/{connector.slug}/')
+ assert 'Method sayHello' in response
+
+ @pytest.mark.parametrize('soap', [BrokenSOAP12], indirect=True)
+ def test_homepage_broken_wsdl(self, app, connector, soap):
+ response = app.get(f'/soap/{connector.slug}/')
+ response = app.get(f'/soap/{connector.slug}/')
+ assert response.pyquery('.down')
+
+
+@pytest.fixture
+def connector(db, soap):
+ return utils.setup_access_rights(
+ SOAPConnector.objects.create(
+ slug='test', wsdl_url=soap.WSDL_URL, zeep_strict=True, zeep_xsd_ignore_sequence_order=False
+ )
)
- mocked_post.return_value = response
-
- soap_resource = SOAPResource()
- client = SOAPClient(soap_resource)
- match = "Unexpected element %s, expected %s" % (repr('price'), repr('skipMe'))
- with pytest.raises(XMLParseError, match=match):
- client.service.GetLastTradePrice(tickerSymbol='banana')
-
- client = SOAPClient(soap_resource, settings=Settings(strict=False))
- result = client.service.GetLastTradePrice(tickerSymbol='banana')
- assert len(result) == 2
- assert result['skipMe'] is None
- assert result['price'] == 4.2
-@mock.patch('requests.sessions.Session.post')
-def test_remove_first_bytes_for_xml(mocked_post):
- response = requests.Response()
- response.status_code = 200
- response._content = force_bytes(
- '''blabla \n
-
-
-
- 1.2
- 4.20
-
-
-\n bloublou'''
- )
- mocked_post.return_value = response
+def test_schemas(connector, soap):
+ assert list(connector.operations_and_schemas) == [('sayHello', soap.INPUT_SCHEMA, soap.OUTPUT_SCHEMA)]
- soap_resource = SOAPResource()
- client = SOAPClient(soap_resource)
- with pytest.raises(TransportError):
- client.service.GetLastTradePrice(tickerSymbol='banana')
+def test_say_hello_method_validation_error(connector, app):
+ resp = app.get('/soap/test/method/sayHello/', status=500)
+ assert dict(resp.json, err_desc=None) == {
+ 'err': 1,
+ 'err_class': 'zeep.exceptions.ValidationError',
+ 'err_desc': None,
+ 'data': None,
+ }
- client = SOAPClient(soap_resource, transport_kwargs={'remove_first_bytes_for_xml': True})
- result = client.service.GetLastTradePrice(tickerSymbol='banana')
- assert len(result) == 2
- assert result['skipMe'] == 1.2
- assert result['price'] == 4.2
+
+def test_say_hello_method_ok_get(connector, app, caplog, soap):
+ resp = app.get('/soap/test/method/sayHello/?' + urllib.parse.urlencode(soap.INPUT_DATA))
+ assert '>John<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()
+ assert '>Bill<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()
+
+ assert '>Doe<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()
+ assert resp.json == {'data': soap.OUTPUT_DATA, 'err': 0}
+
+
+def test_say_hello_method_ok_post_json(connector, app, caplog, soap):
+ resp = app.post_json('/soap/test/method/sayHello/', params=soap.INPUT_DATA)
+ assert '>John<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()
+ assert '>Bill<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()
+ assert '>Doe<' in soap.endpoint_mock.handlers[0].call['requests'][-1].body.decode()
+ assert resp.json == {'data': soap.OUTPUT_DATA, 'err': 0}
+
+
+@pytest.mark.parametrize('soap', [SOAP12], indirect=True)
+class TestAuthencation:
+ def test_basic_auth(self, connector, app, caplog, soap):
+ connector.basic_auth_username = 'username'
+ connector.basic_auth_password = 'password'
+ connector.save()
+
+ app.post_json('/soap/test/method/sayHello/', params=soap.INPUT_DATA)
+ assert (
+ base64.b64decode(
+ soap.endpoint_mock.handlers[0].call['requests'][1].headers['Authorization'].split()[1]
+ )
+ == b'username:password'
+ )
+ assert b'wsse:UsernameToken' not in soap.endpoint_mock.handlers[0].call['requests'][1].body
+
+ def test_username_token(self, connector, app, caplog, soap):
+ connector.zeep_wsse_username = 'username'
+ connector.zeep_wsse_password = 'password'
+ connector.save()
+
+ app.post_json('/soap/test/method/sayHello/', params=soap.INPUT_DATA)
+ assert 'Authorization' not in soap.endpoint_mock.handlers[0].call['requests'][1].headers
+ assert b'wsse:UsernameToken' in soap.endpoint_mock.handlers[0].call['requests'][1].body
diff --git a/tests/test_utils_soap.py b/tests/test_utils_soap.py
new file mode 100644
index 00000000..35fc9472
--- /dev/null
+++ b/tests/test_utils_soap.py
@@ -0,0 +1,109 @@
+# Copyright (C) 2021 Entr'ouvert
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see .
+
+import mock
+import pytest
+import requests
+from django.utils.encoding import force_bytes
+from zeep import Settings
+from zeep.exceptions import TransportError, XMLParseError
+from zeep.plugins import Plugin
+
+from passerelle.utils.soap import SOAPClient
+
+WSDL = 'tests/data/soap.wsdl'
+
+
+class FooPlugin(Plugin):
+ pass
+
+
+class BarPlugin(Plugin):
+ pass
+
+
+class SOAPResource:
+ def __init__(self):
+ self.requests = requests.Session()
+ self.wsdl_url = WSDL
+
+
+def test_soap_client():
+ soap_resource = SOAPResource()
+ plugins = [FooPlugin, BarPlugin]
+ client = SOAPClient(soap_resource, plugins=plugins)
+ assert client.wsdl.location.endswith(WSDL)
+ assert client.transport.session == soap_resource.requests
+ assert client.transport.cache
+ assert client.plugins == plugins
+
+
+@mock.patch('requests.sessions.Session.post')
+def test_disable_strict_mode(mocked_post):
+ response = requests.Response()
+ response.status_code = 200
+ response._content = force_bytes(
+ '''
+
+
+
+ 4.20
+
+
+'''
+ )
+ mocked_post.return_value = response
+
+ soap_resource = SOAPResource()
+ client = SOAPClient(soap_resource)
+ match = "Unexpected element %s, expected %s" % (repr('price'), repr('skipMe'))
+ with pytest.raises(XMLParseError, match=match):
+ client.service.GetLastTradePrice(tickerSymbol='banana')
+
+ client = SOAPClient(soap_resource, settings=Settings(strict=False))
+ result = client.service.GetLastTradePrice(tickerSymbol='banana')
+ assert len(result) == 2
+ assert result['skipMe'] is None
+ assert result['price'] == 4.2
+
+
+@mock.patch('requests.sessions.Session.post')
+def test_remove_first_bytes_for_xml(mocked_post):
+ response = requests.Response()
+ response.status_code = 200
+ response._content = force_bytes(
+ '''blabla \n
+
+
+
+ 1.2
+ 4.20
+
+
+\n bloublou'''
+ )
+ mocked_post.return_value = response
+
+ soap_resource = SOAPResource()
+
+ client = SOAPClient(soap_resource)
+ with pytest.raises(TransportError):
+ client.service.GetLastTradePrice(tickerSymbol='banana')
+
+ client = SOAPClient(soap_resource, transport_kwargs={'remove_first_bytes_for_xml': True})
+ result = client.service.GetLastTradePrice(tickerSymbol='banana')
+ assert len(result) == 2
+ assert result['skipMe'] == 1.2
+ assert result['price'] == 4.2