110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
# 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 <http://www.gnu.org/licenses/>.
|
|
|
|
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(
|
|
'''<?xml version='1.0' encoding='utf-8'?>
|
|
<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<soap-env:Body>
|
|
<ns0:TradePrice xmlns:ns0="http://example.com/stockquote.xsd">
|
|
<price>4.20</price>
|
|
</ns0:TradePrice>
|
|
</soap-env:Body>
|
|
</soap-env:Envelope>'''
|
|
)
|
|
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<?xml version='1.0' encoding='utf-8'?>
|
|
<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<soap-env:Body>
|
|
<ns0:TradePrice xmlns:ns0="http://example.com/stockquote.xsd">
|
|
<skipMe>1.2</skipMe>
|
|
<price>4.20</price>
|
|
</ns0:TradePrice>
|
|
</soap-env:Body>
|
|
</soap-env:Envelope>\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
|