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