1#!/usr/bin/env python 2# 3# spyne - Copyright (C) spyne contributors. 4# 5# This library is free software; you can redistribute it and/or 6# modify it under the terms of the GNU Lesser General Public 7# License as published by the Free Software Foundation; either 8# version 2.1 of the License, or (at your option) any later version. 9# 10# This library is distributed in the hope that it will be useful, 11# but WITHOUT ANY WARRANTY; without even the implied warranty of 12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 13# Lesser General Public License for more details. 14# 15# You should have received a copy of the GNU Lesser General Public 16# License along with this library; if not, write to the Free Software 17# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 18# 19 20 21import logging 22logging.basicConfig(level=logging.DEBUG) 23 24import unittest 25 26from lxml import etree 27 28from spyne.application import Application 29 30from spyne.test.interface.wsdl import AppTestWrapper 31from spyne.test.interface.wsdl import build_app 32from spyne.test.interface.wsdl.defult_services import TDefaultPortService 33from spyne.test.interface.wsdl.defult_services import \ 34 TDefaultPortServiceMultipleMethods 35 36from spyne.const import REQUEST_SUFFIX 37from spyne.const import RESPONSE_SUFFIX 38from spyne.const import ARRAY_SUFFIX 39 40from spyne.decorator import srpc 41from spyne.service import Service 42from spyne.interface.wsdl import Wsdl11 43from spyne.model.complex import Array 44from spyne.model.primitive import String 45 46ns = { 47 'wsdl': 'http://schemas.xmlsoap.org/wsdl/', 48 'xs': 'http://www.w3.org/2001/XMLSchema', 49} 50 51 52class TestDefaultWSDLBehavior(unittest.TestCase): 53 def _default_service(self, app_wrapper, service_name): 54 self.assertEqual(1, len(app_wrapper.get_service_list())) 55 56 services = app_wrapper.get_service_list() 57 service = services[0] 58 59 # the default behavior requires that there be only a single service 60 self.assertEqual(1, len(services)) 61 self.assertEqual(service_name, service.get('name')) 62 63 # Test the default service has the correct number of ports 64 # the default behavior requires that there be only a single port 65 ports = app_wrapper.get_port_list(service) 66 self.assertEqual(len(ports), 1) 67 68 def _default_port_type(self, app_wrapper, portType_name, op_count): 69 # Verify the portType Count 70 portTypes = app_wrapper.get_port_types() 71 72 # there should be only one portType 73 self.assertEqual(1, len(portTypes)) 74 75 # Verify the portType name 76 portType = portTypes[0] 77 # Check the name of the port 78 self.assertEqual(portType_name, portType.get('name')) 79 80 # verify that the portType definition has the correct 81 # number of operations 82 ops = app_wrapper.get_port_operations(portType) 83 self.assertEqual(op_count, len(ops)) 84 85 def _default_binding(self, wrapper, binding_name, opp_count): 86 # the default behavior is only single binding 87 bindings = wrapper.get_bindings() 88 self.assertEqual(1, len(bindings)) 89 90 # check for the correct binding name 91 binding = bindings[0] 92 name = binding.get('name') 93 self.assertEqual(binding_name, name) 94 95 # Test that the default service contains the soap binding 96 sb = wrapper.get_soap_bindings(binding) 97 self.assertEqual(1, len(sb)) 98 99 # verify the correct number of operations 100 ops = wrapper.get_binding_operations(binding) 101 self.assertEqual(opp_count, len(ops)) 102 103 def _default_binding_methods(self, wrapper, op_count, op_names): 104 binding = wrapper.get_bindings()[0] 105 operations = wrapper.get_binding_operations(binding) 106 107 # Check the number of operations bound to the port 108 self.assertEqual(op_count, len(operations)) 109 110 # Check the operation names are correct 111 for op in operations: 112 self.assertTrue(op.get('name') in op_names) 113 114 def test_default_port_type(self): 115 # Test the default port is created 116 # Test the default port has the correct name 117 app = build_app( 118 [TDefaultPortService()], 119 'DefaultPortTest', 120 'DefaultPortName' 121 ) 122 123 wrapper = AppTestWrapper(app) 124 self._default_port_type(wrapper, 'DefaultPortName', 1) 125 126 def test_default_port_type_multiple(self): 127 app = build_app( 128 [TDefaultPortServiceMultipleMethods()], 129 'DefaultServiceTns', 130 'MultipleDefaultPortServiceApp' 131 ) 132 133 wrapper = AppTestWrapper(app) 134 135 self._default_port_type(wrapper, "MultipleDefaultPortServiceApp", 3) 136 137 def test_default_binding(self): 138 app = build_app( 139 [TDefaultPortService()], 140 'DefaultPortTest', 141 'DefaultBindingName' 142 ) 143 144 wrapper = AppTestWrapper(app) 145 146 self._default_binding(wrapper, "DefaultBindingName", 1) 147 148 def test_default_binding_multiple(self): 149 app = build_app( 150 [TDefaultPortServiceMultipleMethods()], 151 'DefaultPortTest', 152 'MultipleDefaultBindingNameApp' 153 ) 154 155 wrapper = AppTestWrapper(app) 156 157 self._default_binding(wrapper, 'MultipleDefaultBindingNameApp', 3) 158 159 def test_default_binding_methods(self): 160 app = build_app( 161 [TDefaultPortService()], 162 'DefaultPortTest', 163 'DefaultPortMethods' 164 ) 165 166 wrapper = AppTestWrapper(app) 167 168 self._default_binding_methods( 169 wrapper, 170 1, 171 ['echo_default_port_service'] 172 ) 173 174 def test_bare_simple(self): 175 class SomeService(Service): 176 @srpc(String, _returns=String, _body_style='bare') 177 def whatever(ss): 178 return ss 179 180 app = Application([SomeService], tns='tns') 181 app.transport = 'None' 182 183 wsdl = Wsdl11(app.interface) 184 wsdl.build_interface_document('url') 185 wsdl = etree.fromstring(wsdl.get_interface_document()) 186 187 schema = wsdl.xpath( 188 '/wsdl:definitions/wsdl:types/xs:schema[@targetNamespace="tns"]', 189 namespaces=ns, 190 ) 191 assert len(schema) == 1 192 193 print(etree.tostring(wsdl, pretty_print=True)) 194 195 elts = schema[0].xpath( 196 'xs:element[@name="whatever%s"]' % REQUEST_SUFFIX, namespaces=ns) 197 assert len(elts) > 0 198 assert elts[0].attrib['type'] == 'xs:string' 199 200 elts = schema[0].xpath( 201 'xs:element[@name="whatever%s"]' % RESPONSE_SUFFIX, namespaces=ns) 202 assert len(elts) > 0 203 assert elts[0].attrib['type'] == 'xs:string' 204 205 def test_bare_with_conflicting_types(self): 206 class SomeService(Service): 207 @srpc(Array(String), _returns=Array(String)) 208 def whatever(sa): 209 return sa 210 211 @srpc(Array(String), _returns=Array(String), _body_style='bare') 212 def whatever_bare(sa): 213 return sa 214 215 app = Application([SomeService], tns='tns') 216 app.transport = 'None' 217 218 wsdl = Wsdl11(app.interface) 219 wsdl.build_interface_document('url') 220 wsdl = etree.fromstring(wsdl.get_interface_document()) 221 schema, = wsdl.xpath( 222 '/wsdl:definitions/wsdl:types/xs:schema[@targetNamespace="tns"]', 223 namespaces=ns, 224 ) 225 226 print(etree.tostring(schema, pretty_print=True)) 227 228 assert len(schema.xpath( 229 'xs:complexType[@name="string%s"]' % ARRAY_SUFFIX, 230 namespaces=ns)) > 0 231 232 elts = schema.xpath( 233 'xs:element[@name="whatever_bare%s"]' % REQUEST_SUFFIX, 234 namespaces=ns) 235 236 assert len(elts) > 0 237 assert elts[0].attrib['type'] == 'tns:string%s' % ARRAY_SUFFIX 238 239 elts = schema.xpath( 240 'xs:element[@name="whatever_bare%s"]' % RESPONSE_SUFFIX, 241 namespaces=ns) 242 243 assert len(elts) > 0 244 assert elts[0].attrib['type'] == 'tns:string%s' % ARRAY_SUFFIX 245 246 247if __name__ == '__main__': 248 unittest.main() 249