362 lines
12 KiB
Python
362 lines
12 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 base64
|
|
import urllib.parse
|
|
|
|
import pytest
|
|
|
|
from passerelle.apps.soap.models import SOAPConnector
|
|
|
|
from . import utils
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
class SOAP11:
|
|
VERSION = '1.1'
|
|
ENDPOINT_URL = 'https://www.examples.com/SayHello/'
|
|
WSDL_CONTENT = '''\
|
|
<definitions name = "HelloService"
|
|
targetNamespace = "http://www.examples.com/wsdl/HelloService.wsdl"
|
|
xmlns = "http://schemas.xmlsoap.org/wsdl/"
|
|
xmlns:soap = "http://schemas.xmlsoap.org/wsdl/soap/"
|
|
xmlns:tns = "http://www.examples.com/wsdl/HelloService.wsdl"
|
|
xmlns:xsd = "http://www.w3.org/2001/XMLSchema">
|
|
|
|
<types>
|
|
<schema targetNamespace="http://www.examples.com/wsdl/HelloService.wsdl" xmlns="http://www.w3.org/2001/XMLSchema">
|
|
<element name="firstName">
|
|
<complexType name="listofstring">
|
|
<sequence>
|
|
<element name="string" type="string" maxOccurs="unbounded"/>
|
|
</sequence>
|
|
</complexType>
|
|
</element>
|
|
</schema>
|
|
</types>
|
|
|
|
<message name = "SayHelloRequest">
|
|
<part name = "firstName" element="tns:firstName"/>
|
|
<part name = "lastName" type = "xsd:string"/>
|
|
</message>
|
|
|
|
<message name = "SayHelloResponse">
|
|
<part name = "greeting" type = "xsd:string"/>
|
|
<part name = "who" type = "xsd:string"/>
|
|
</message>
|
|
|
|
<portType name = "Hello_PortType">
|
|
<operation name = "sayHello">
|
|
<input message = "tns:SayHelloRequest"/>
|
|
<output message = "tns:SayHelloResponse"/>
|
|
</operation>
|
|
</portType>
|
|
|
|
<binding name = "Hello_Binding" type = "tns:Hello_PortType">
|
|
<soap:binding style = "rpc"
|
|
transport = "http://schemas.xmlsoap.org/soap/http"/>
|
|
<operation name = "sayHello">
|
|
<soap:operation soapAction = "sayHello"/>
|
|
<input>
|
|
<soap:body
|
|
encodingStyle = "http://schemas.xmlsoap.org/soap/encoding/"
|
|
namespace = "urn:examples:helloservice"
|
|
use = "encoded"/>
|
|
</input>
|
|
|
|
<output>
|
|
<soap:body
|
|
encodingStyle = "http://schemas.xmlsoap.org/soap/encoding/"
|
|
namespace = "urn:examples:helloservice"
|
|
use = "encoded"/>
|
|
</output>
|
|
</operation>
|
|
</binding>
|
|
|
|
<service name = "Hello_Service">
|
|
<documentation>WSDL File for HelloService</documentation>
|
|
<port binding = "tns:Hello_Binding" name = "Hello_Port">
|
|
<soap:address
|
|
location = "http://www.examples.com/SayHello/" />
|
|
</port>
|
|
</service>
|
|
</definitions>'''
|
|
WSDL_URL = 'https://example.com/service.wsdl'
|
|
SOAP_RESPONSE = '''\
|
|
<?xml version="1.0" encoding="utf-8"?>
|
|
<soap:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
|
|
xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<soap:Body>
|
|
<SayHelloResponse xmlns="urn:examples:helloservice">
|
|
<greeting>Hello</greeting>
|
|
<who>John!</who>
|
|
</SayHelloResponse>
|
|
</soap:Body>
|
|
</soap:Envelope>'''
|
|
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!',
|
|
}
|
|
VALIDATION_ERROR = 'Missing element firstName (sayHello.firstName)'
|
|
|
|
|
|
class SOAP12(SOAP11):
|
|
VERSION = '1.2'
|
|
ENDPOINT_URL = 'https://www.examples.com/SayHello/'
|
|
WSDL_CONTENT = f'''\
|
|
<?xml version="1.0"?>
|
|
<wsdl:definitions xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"
|
|
xmlns:soap12="http://schemas.xmlsoap.org/wsdl/soap12/"
|
|
xmlns:tns="urn:examples:helloservice"
|
|
targetNamespace="urn:examples:helloservice">
|
|
|
|
<wsdl:types>
|
|
<xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema"
|
|
xmlns:tns="urn:examples:helloservice"
|
|
targetNamespace="urn:examples:helloservice">
|
|
<xsd:element name="sayHello">
|
|
<xsd:complexType>
|
|
<xsd:sequence>
|
|
<xsd:element name="firstName" type="xsd:string" maxOccurs="unbounded"/>
|
|
<xsd:element name="lastName" type="xsd:string"/>
|
|
</xsd:sequence>
|
|
</xsd:complexType>
|
|
</xsd:element>
|
|
<xsd:element name="sayHelloResponse">
|
|
<xsd:complexType>
|
|
<xsd:sequence>
|
|
<xsd:element name="greeting" type="xsd:string"/>
|
|
<xsd:element name="who" type="xsd:string" maxOccurs="unbounded"/>
|
|
</xsd:sequence>
|
|
</xsd:complexType>
|
|
</xsd:element>
|
|
</xsd:schema>
|
|
</wsdl:types>
|
|
|
|
<wsdl:message name="sayHello">
|
|
<wsdl:part name="sayHelloInputPart" element="tns:sayHello"/>
|
|
</wsdl:message>
|
|
<wsdl:message name="sayHelloResponse">
|
|
<wsdl:part name="sayHelloOutputPart" element="tns:sayHelloResponse"/>
|
|
</wsdl:message>
|
|
|
|
<wsdl:portType name="sayHelloPortType">
|
|
<wsdl:operation name="sayHello">
|
|
<wsdl:input name="sayHello" message="tns:sayHello"/>
|
|
<wsdl:output name="sayHelloResponse" message="tns:sayHelloResponse"/>
|
|
</wsdl:operation>
|
|
</wsdl:portType>
|
|
|
|
<wsdl:binding name="sayHelloBinding"
|
|
type="tns:sayHelloPortType">
|
|
<soap12:binding style="document" transport="http://schemas.xmlsoap.org/soap/http"/>
|
|
<wsdl:operation name="sayHello">
|
|
<soap12:operation style="document"/>
|
|
<wsdl:input name="sayHello">
|
|
<soap12:body use="literal"/>
|
|
</wsdl:input>
|
|
<wsdl:output name="sayHelloResponse">
|
|
<soap12:body use="literal"/>
|
|
</wsdl:output>
|
|
</wsdl:operation>
|
|
</wsdl:binding>
|
|
|
|
<wsdl:service name="sayHelloService">
|
|
<wsdl:port name="sayHelloPort"
|
|
binding="tns:sayHelloBinding">
|
|
<soap12:address location="{ENDPOINT_URL}"/>
|
|
</wsdl:port>
|
|
</wsdl:service>
|
|
</wsdl:definitions>'''
|
|
SOAP_RESPONSE = '''\
|
|
<?xml version="1.0" encoding="UTF-8"?>
|
|
<soap:Envelope
|
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
|
|
xmlns:soap="http://www.w3.org/2003/05/soap-envelope"
|
|
xmlns:ser="http://www.herongyang.com/Service/">
|
|
<soap:Header/>
|
|
<soap:Body>
|
|
<sayHelloResponse xmlns="urn:examples:helloservice">
|
|
<greeting>Hello</greeting>
|
|
<who>John!</who>
|
|
</sayHelloResponse>
|
|
</soap:Body>
|
|
</soap:Envelope>'''
|
|
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!'],
|
|
}
|
|
VALIDATION_ERROR = 'Expected at least 1 items (minOccurs check) 0 items found. (sayHello.firstName)'
|
|
|
|
|
|
class BrokenSOAP12(SOAP12):
|
|
WSDL_CONTENT = SOAP12.WSDL_CONTENT[-100:] # truncate the WSDL to break it
|
|
|
|
|
|
@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
|
|
|
|
|
|
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
|
|
)
|
|
)
|
|
|
|
|
|
def test_schemas(connector, soap):
|
|
assert list(connector.operations_and_schemas) == [('sayHello', soap.INPUT_SCHEMA, soap.OUTPUT_SCHEMA)]
|
|
|
|
|
|
def test_say_hello_method_validation_error(connector, soap, app):
|
|
resp = app.get('/soap/test/method/sayHello/')
|
|
assert resp.json == {
|
|
"err": 1,
|
|
"err_class": "passerelle.utils.soap.SOAPError",
|
|
"err_desc": soap.VALIDATION_ERROR,
|
|
"data": None,
|
|
}
|
|
|
|
|
|
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_get_with_signature(connector, app, caplog, soap):
|
|
resp = app.get('/soap/test/method/sayHello/?' + urllib.parse.urlencode(soap.INPUT_DATA) + '&signature=1')
|
|
assert resp.json['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
|