Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP

Loading…

HttpRpc protocol got some love. #86

Merged
merged 23 commits into from

1 participant

@plq
Owner

No description provided.

@plq plq merged commit 202ebe5 into from
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Commits on Sep 11, 2011
  1. @plq
  2. @plq
  3. @plq
  4. @plq
  5. @plq

    fix soap mime type.

    plq authored
  6. @plq

    hard-code utf8 for soap.

    plq authored
  7. @plq

    XmlObject protocol does not distinguish between have header and body …

    plq authored
    …parts of a message, so move code from decompose_incoming_envelope function to create_in_document
  8. @plq

    test readme tweak.

    plq authored
  9. @plq
  10. @plq

    remove stray import.

    plq authored
  11. @plq
  12. @plq

    improve wording in comment

    plq authored
  13. @plq

    fix service tests.

    plq authored
  14. @plq

    add httprpc tests.

    plq authored
  15. @plq
  16. @plq
  17. @plq
  18. @plq
  19. @plq

    comment tweaks.

    plq authored
  20. @plq
  21. @plq

    add 'wsgi_method_not_found' event.

    plq authored
  22. @plq
  23. @plq

    changelog update

    plq authored
This page is out of date. Refresh to see the latest.
View
12 CHANGELOG.rst
@@ -2,6 +2,18 @@
Changelog
=========
+rpclib-2.3.1-beta
+-----------------
+ * HttpRpc protocol now returns 404 when a requested resource was not found.
+ * New tests added for HttpRpc protocol.
+ * Miscellanous other fixes.
+
+rpclib-2.3.0-beta
+-----------------
+ * Documentation improvements.
+ * rpclib.protocol.xml.XmlObject is now working as out_protocol.
+ * Many fixes.
+
rpclib-2.2.3-beta
------------------
* Documentation improvements.
View
6 src/rpclib/__init__.py
@@ -18,3 +18,9 @@
#
__version__ = '2.3.0-beta'
+
+from _base import TransportContext
+from _base import EventContext
+from _base import MethodContext
+from _base import MethodDescriptor
+from _base import EventManager
View
3  src/rpclib/error.py
@@ -0,0 +1,3 @@
+
+class NotFoundError(Exception):
+ """Raised when the requested resource was not found."""
View
13 src/rpclib/model/complex.py
@@ -24,10 +24,9 @@
import logging
logger = logging.getLogger(__name__)
-from lxml import etree
-
from rpclib.model import ModelBase
from rpclib.model import nillable_dict
+from rpclib.model import nillable_string
from rpclib.util.odict import odict as TypeInfo
from rpclib.const import xml_ns as namespace
@@ -223,6 +222,16 @@ def get_flat_type_info(clz, retval={}):
return retval
@classmethod
+ @nillable_string
+ def to_string(cls, value):
+ raise ValueError("Only primitives can be serialized to string.")
+
+ @classmethod
+ @nillable_string
+ def from_string(cls, string):
+ raise ValueError("Only primitives can be deserialized from string.")
+
+ @classmethod
@nillable_dict
def from_dict(cls, in_dict):
inst = cls.get_deserialization_instance()
View
1  src/rpclib/protocol/__init__.py
@@ -18,4 +18,3 @@
#
from _base import ProtocolBase
-from _base import ValidationError
View
14 src/rpclib/protocol/_base.py
@@ -30,10 +30,8 @@
_ns_xsd = rpclib.const.xml_ns.xsd
from rpclib._base import EventManager
-from rpclib.model.fault import Fault
-
-class ValidationError(Fault):
- pass
+from rpclib.error import NotFoundError
+# from pprint import pformat
class ProtocolBase(object):
"""This is the abstract base class for all protocol implementations. Child
@@ -105,10 +103,10 @@ def set_method_descriptor(self, ctx):
ctx.service_class = self.app.interface.service_mapping.get(name, None)
if ctx.service_class is None:
- logger.debug(self.app.interface.service_mapping.keys())
- raise Exception('Method %r not bound to a service class.' % name)
+ # logger.debug(pformat(self.app.interface.service_mapping.keys()))
+ raise NotFoundError('Method %r not bound to a service class.' % name)
ctx.descriptor = ctx.app.interface.method_mapping.get(name, None)
if ctx.descriptor is None:
- logger.debug(ctx.app.interface.method_mapping.keys())
- raise Exception('Method %r not found.' % name)
+ # logger.debug(pformat(ctx.app.interface.method_mapping.keys()))
+ raise NotFoundError('Method %r not found.' % name)
View
46 src/rpclib/protocol/http.py
@@ -22,10 +22,23 @@
import logging
logger = logging.getLogger(__name__)
-from rpclib.protocol import ProtocolBase
import urlparse
+from rpclib.error import NotFoundError
+from rpclib.server.wsgi import HTTP_404
+from rpclib.protocol import ProtocolBase
+
# this is not exactly rest, because it ignores http verbs.
+
+def _get_http_headers(req_env):
+ retval = {}
+
+ for k,v in req_env.iteritems():
+ if k.startswith("HTTP_"):
+ retval[k[5:].lower()]= v
+
+ return retval
+
class HttpRpc(ProtocolBase):
"""The so-called ReST-minus-the-verbs HttpRpc protocol implementation.
It only works with the http server (wsgi) transport.
@@ -42,9 +55,13 @@ def create_in_document(self, ctx, in_string_encoding=None):
ctx.transport.req_env['PATH_INFO'].split('/')[-1])
logger.debug("\033[92mMethod name: %r\033[0m" % ctx.method_request_string)
- self.app.in_protocol.set_method_descriptor(ctx)
+ try:
+ self.app.in_protocol.set_method_descriptor(ctx)
+ except NotFoundError, e:
+ ctx.transport.resp_code = HTTP_404
+ raise
- ctx.in_header_doc = None
+ ctx.in_header_doc = _get_http_headers(ctx.transport.req_env)
ctx.in_body_doc = urlparse.parse_qs(ctx.transport.req_env['QUERY_STRING'])
logger.debug(repr(ctx.in_body_doc))
@@ -62,25 +79,20 @@ def deserialize(self, ctx):
def serialize(self, ctx):
result_message_class = ctx.descriptor.out_message
- result_message = result_message_class()
# assign raw result to its wrapper, result_message
out_type_info = result_message_class._type_info
- if len(out_type_info) > 0:
- if len(out_type_info) == 1:
- attr_name = result_message_class._type_info.keys()[0]
- setattr(result_message, attr_name, ctx.out_object)
-
- else:
- for i in range(len(out_type_info)):
- attr_name=result_message_class._type_info.keys()[i]
- setattr(result_message, attr_name, ctx.out_object[i])
+ if len(out_type_info) == 1:
+ out_class = out_type_info.values()[0]
+ if ctx.out_object is None:
+ ctx.out_document = u''
+ else:
+ ctx.out_document = out_class.to_string(ctx.out_object)
- wrapped_result = result_message_class.to_dict(result_message)
-
- ctx.out_document, = wrapped_result.itervalues()
+ else:
+ raise ValueError("HttpRpc protocol can only serialize primitives.")
self.event_manager.fire_event('serialize', ctx)
def create_out_string(self, ctx, out_string_encoding=None):
- ctx.out_string = ctx.out_document
+ ctx.out_string = [ctx.out_document.encode('utf8')]
View
7 src/rpclib/protocol/soap/soap11.py
@@ -34,7 +34,6 @@
from rpclib.protocol.soap.mime import collapse_swa
from rpclib.model.fault import Fault
-from rpclib.model.primitive import string_encoding
class ValidationError(Fault):
pass
@@ -71,7 +70,7 @@ def _from_soap(in_envelope_xml, xmlids=None):
def _parse_xml_string(xml_string, charset=None):
try:
if charset is None:
- charset = string_encoding
+ charset = 'utf-8'
root, xmlids = etree.XMLID(xml_string.decode(charset))
@@ -130,7 +129,7 @@ class OUT_WRAPPER:
pass
allowed_http_verbs = ['POST']
- mime_type = 'application/soap+xml'
+ mime_type = 'text/xml; charset=utf-8'
def __init__(self, app=None):
XmlObject.__init__(self, app)
@@ -149,7 +148,7 @@ def create_in_document(self, ctx, charset=None):
def create_out_string(self, ctx, charset=None):
"""Sets an iterable of string fragments to ctx.out_string"""
if charset is None:
- charset = string_encoding
+ charset = 'utf-8'
ctx.out_string = [etree.tostring(ctx.out_document, xml_declaration=True,
encoding=charset)]
View
22 src/rpclib/protocol/xml/_base.py
@@ -26,6 +26,7 @@
from rpclib.util.cdict import cdict
+from rpclib.error import NotFoundError
from rpclib.model import ModelBase
from rpclib.model.complex import Array
@@ -82,10 +83,6 @@
EnumBase: enum_from_element,
})
-class NotFoundError(Exception):
- """Raised when the requested resource was not found."""
- pass
-
class XmlObject(ProtocolBase):
def from_element(self, cls, element):
handler = _deserialization_handlers[cls]
@@ -98,15 +95,6 @@ def to_parent_element(self, cls, value, tns, parent_elt, * args, ** kwargs):
def create_in_document(self, ctx, charset=None):
ctx.in_document = etree.fromstring(ctx.in_string, charset)
- def create_out_string(self, ctx, charset=None):
- """Sets an iterable of string fragments to ctx.out_string"""
- if charset is None:
- charset = 'utf8'
-
- ctx.out_string = [etree.tostring(ctx.out_document, xml_declaration=True,
- encoding=charset)]
-
- def decompose_incoming_envelope(self, ctx):
body_doc = ctx.in_document
try:
@@ -146,6 +134,14 @@ def decompose_incoming_envelope(self, ctx):
# payload. That's SOAP's job to do.
ctx.in_body_doc = body_doc
+ def create_out_string(self, ctx, charset=None):
+ """Sets an iterable of string fragments to ctx.out_string"""
+ if charset is None:
+ charset = 'utf8'
+
+ ctx.out_string = [etree.tostring(ctx.out_document, xml_declaration=True,
+ encoding=charset)]
+
def deserialize(self, ctx, way='out'):
"""Takes a MethodContext instance and a string containing ONE root xml
tag.
View
9 src/rpclib/server/_base.py
@@ -26,12 +26,6 @@
from rpclib.model.fault import Fault
from rpclib._base import EventManager
-class InternalError(Fault):
- pass
-
-class ValidationError(Fault):
- pass
-
class ServerBase(object):
"""This class is the abstract base class for all server transport
implementations. Unlike the client transports, this class does not define
@@ -102,3 +96,6 @@ def get_out_string(self, ctx):
else:
ctx.service_class.event_manager.fire_event(
'method_exception_string', ctx)
+
+ if ctx.out_string is None:
+ ctx.out_string = [""]
View
76 src/rpclib/server/wsgi.py
@@ -26,8 +26,10 @@
import cgi
-from rpclib._base import MethodContext
-from rpclib.model.fault import Fault
+from rpclib import TransportContext
+from rpclib import MethodContext
+
+from rpclib.error import NotFoundError
from rpclib.protocol.soap.mime import apply_mtom
from rpclib.util import reconstruct_url
from rpclib.server import ServerBase
@@ -59,18 +61,25 @@ def reconstruct_wsgi_request(http_env):
return input.read(length), charset
-class WsgiMethodContext(MethodContext):
- def __init__(self, app, req_env, content_type):
- MethodContext.__init__(self, app)
+class WsgiTransportContext(TransportContext):
+ def __init__(self, req_env, content_type):
+ TransportContext.__init__(self, 'wsgi')
- self.transport.type = 'wsgi'
- self.transport.req_env = req_env
- self.transport.resp_headers = {
+ self.req_env = req_env
+ self.resp_headers = {
'Content-Type': content_type,
'Content-Length': '0',
}
- self.transport.req_method = req_env.get('REQUEST_METHOD', None)
- self.transport.wsdl_error = None
+ self.resp_code = None
+ self.req_method = req_env.get('REQUEST_METHOD', None)
+ self.wsdl_error = None
+
+
+class WsgiMethodContext(MethodContext):
+ def __init__(self, app, req_env, content_type):
+ MethodContext.__init__(self, app)
+
+ self.transport = WsgiTransportContext(req_env, content_type)
class WsgiApplication(ServerBase):
@@ -108,7 +117,7 @@ def __call__(self, req_env, start_response, wsgi_url=None):
def __is_wsdl_request(self, req_env):
# Get the wsdl for the service. Assume path_info matches pattern:
# /stuff/stuff/stuff/serviceName.wsdl or
- # /stuff/stuff/stuff/serviceName/?wsdl
+ # /stuff/stuff/stuff/serviceName/?wsdl with anything between ? and wsdl.
return (
req_env['REQUEST_METHOD'].lower() == 'get'
@@ -119,7 +128,7 @@ def __is_wsdl_request(self, req_env):
)
def __handle_wsdl_request(self, req_env, start_response, url):
- ctx = WsgiMethodContext(self.app, req_env, 'text/xml')
+ ctx = WsgiMethodContext(self.app, req_env, 'text/xml; charset=utf-8')
try:
wsdl = self.app.interface.get_interface_document()
@@ -155,34 +164,45 @@ def __handle_rpc(self, req_env, start_response):
ctx.in_string, in_string_charset = reconstruct_wsgi_request(req_env)
- self.get_in_object(ctx, in_string_charset)
+ try:
+ self.get_in_object(ctx, in_string_charset)
+ except NotFoundError, e:
+ pass
- return_code = HTTP_200
if ctx.in_error:
out_object = ctx.in_error
- return_code = HTTP_500
+ if ctx.transport.resp_code is None:
+ ctx.transport.resp_code = HTTP_500
else:
if ctx.service_class == None:
- start_response(HTTP_404, ctx.transport.resp_headers.items())
- return ['']
+ if ctx.transport.resp_code is None:
+ ctx.transport.resp_code = HTTP_500
+
+ ctx.out_string = [ctx.transport.resp_code]
+
+ self.event_manager.fire_event('wsgi_method_not_found', ctx)
+
+ start_response(ctx.transport.resp_code, ctx.transport.resp_headers.items())
+ return ctx.out_string
self.get_out_object(ctx)
- if not (ctx.out_error is None):
- return_code = HTTP_500
+ if ctx.out_error is None:
+ ctx.transport.resp_code = HTTP_200
+ else:
+ ctx.transport.resp_code = HTTP_500
self.get_out_string(ctx)
- if ctx.out_string is None:
- ctx.out_string = [""]
# implementation hook
self.event_manager.fire_event('wsgi_return', ctx)
if ctx.descriptor and ctx.descriptor.mtom:
- # when there are more than one return type, the result is
+ # when there is more than one return type, the result is
# encapsulated inside a list. when there's just one, the result
- # is returned unencapsulated. the apply_mtom always expects the
- # objects to be inside an iterable, hence the following test.
+ # is returned in a non-encapsulated form. the apply_mtom always
+ # expects the objects to be inside an iterable, hence the following
+ # test.
out_type_info = ctx.descriptor.out_message._type_info
if len(out_type_info) == 1:
out_object = [out_object]
@@ -193,8 +213,12 @@ def __handle_rpc(self, req_env, start_response):
out_object
)
- # initiate the response
+ # We can't set the content-length if we want to support any kind of
+ # python iterable as output. We can't iterate and count, that defeats
+ # the whole point.
del ctx.transport.resp_headers['Content-Length']
- start_response(return_code, ctx.transport.resp_headers.items())
+
+ # initiate the response
+ start_response(ctx.transport.resp_code, ctx.transport.resp_headers.items())
return ctx.out_string
View
29 src/rpclib/test/README
@@ -12,13 +12,23 @@ detect such corner cases is to have a great test suite.
Requirements
------------
-While the stock unittest package is normally enough to run Python tests, using
-py.test from pytest package is just a more pleasant way to run them. Simply
-easy_install pytest to get it. You can run the following command in the test
-directory: ::
+While simply executing test modules is normally enough to run Python tests,
+using py.test from pytest package is just a more pleasant way to run them.
+Simply easy_install pytest to get it. You can run the following command in the
+test directory: ::
py.test -v --tb=short
+You can use the module name as an argument: ::
+
+ py.test -v --tb=short test_sqla.py
+
+You can also choose which test to run: ::
+
+ py.test -v --tb=short test_sqla.py -k test_same_table_inheritance
+
+See [pytest documentation](http://pytest.org/latest/) for more info.
+
Note that you need to do several other preparations to have the interop tests
working. See the next section for the specifics.
@@ -28,6 +38,15 @@ Interoperability Tests
The interoperability servers require twisted.web.
+.Python
+^^^^^^^
+
+Python interop tests currently use Rpclib's own clients and suds. The suds test
+is the first thing we check and try not to break.
+
+Two tests that fail in the suds interop tests due to the lack of proper assert
+statements, so they're false alarms.
+
.Net
^^^^
@@ -78,5 +97,3 @@ Here's the directory tree from a working setup:
| |-- README.txt
| `-- (...)
`-- (...)
-
-
View
2  src/rpclib/test/interop/server/httprpc_pod_basic.py
@@ -41,7 +41,7 @@
wsgi_application = WsgiApplication(httprpc_soap_application)
server = make_server('0.0.0.0', 9757, validator(wsgi_application))
- logger.info('Starting interop server at %s:%s.' % ('0.0.0.0', 9756))
+ logger.info('Starting interop server at %s:%s.' % ('0.0.0.0', 9757))
logger.info('WSDL is at: /?wsdl')
server.serve_forever()
View
86 src/rpclib/test/interop/test_httprpc.py
@@ -0,0 +1,86 @@
+#!/usr/bin/env python
+#
+# rpclib - Copyright (C) Rpclib contributors.
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library 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
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301
+#
+
+import pytz
+import unittest
+import urllib
+import urllib2
+
+from datetime import datetime
+
+class TestHttpRpc(unittest.TestCase):
+ def test_404(self):
+ url = 'http://localhost:9757/404'
+ try:
+ data = urllib2.urlopen(url).read()
+ except urllib2.HTTPError, e:
+ assert e.code == 404
+
+ def test_500(self):
+ url = 'http://localhost:9757/python_exception'
+ try:
+ data = urllib2.urlopen(url).read()
+ except urllib2.HTTPError, e:
+ assert e.code == 500
+
+ def test_500_2(self):
+ url = 'http://localhost:9757/soap_exception'
+ try:
+ data = urllib2.urlopen(url).read()
+ except urllib2.HTTPError, e:
+ assert e.code == 500
+
+ def test_echo_string(self):
+ url = 'http://localhost:9757/echo_string?s=punk'
+ data = urllib2.urlopen(url).read()
+
+ assert data == 'punk'
+
+ def test_echo_integer(self):
+ url = 'http://localhost:9757/echo_integer?i=444'
+ data = urllib2.urlopen(url).read()
+
+ assert data == '444'
+
+ def test_echo_datetime(self):
+ dt = datetime.now().isoformat()
+ params = urllib.urlencode({
+ 'dt': dt,
+ })
+
+ print params
+ url = 'http://localhost:9757/echo_datetime?%s' % str(params)
+ data = urllib2.urlopen(url).read()
+
+ assert dt == data
+
+ def test_echo_datetime_tz(self):
+ dt = datetime.now(pytz.utc).isoformat()
+ params = urllib.urlencode({
+ 'dt': dt,
+ })
+
+ print params
+ url = 'http://localhost:9757/echo_datetime?%s' % str(params)
+ data = urllib2.urlopen(url).read()
+
+ assert dt == data
+
+if __name__ == '__main__':
+ unittest.main()
View
5 src/rpclib/test/interop/test_suds.py
@@ -62,10 +62,11 @@ def test_validation(self):
non_nillable_class.s = None
try:
- ret = self.client.service.non_nillable(non_nillable_class)
- raise Exception("must fail")
+ self.client.service.non_nillable(non_nillable_class)
except WebFault, e:
pass
+ else:
+ raise Exception("must fail")
def test_echo_integer_array(self):
ia = self.client.factory.create('integerArray')
View
104 src/rpclib/test/protocol/test_http.py
@@ -0,0 +1,104 @@
+
+#!/usr/bin/env python
+#
+# rpclib - Copyright (C) Rpclib contributors.
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library 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
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301
+#
+
+import logging
+logging.basicConfig(level=logging.DEBUG)
+
+import unittest
+
+from rpclib.application import Application
+from rpclib.decorator import srpc
+from rpclib.model.primitive import Integer
+from rpclib.model.primitive import String
+from rpclib.model.complex import ComplexModel
+from rpclib.interface.wsdl import Wsdl11
+from rpclib.protocol.http import HttpRpc
+from rpclib.protocol.soap import Soap11
+from rpclib.service import ServiceBase
+
+class Test(unittest.TestCase):
+ '''Most of the service tests are performed through the interop tests.'''
+
+ def test_multiple_return(self):
+ class SomeNotSoComplexModel(ComplexModel):
+ s = String
+
+ class SomeService(ServiceBase):
+ @srpc(_returns=[Integer, String])
+ def some_call():
+ return 1,'s'
+
+ app = Application([SomeService], 'tns', Wsdl11(), HttpRpc(), HttpRpc())
+
+ from rpclib.server.wsgi import WsgiMethodContext
+
+ ctx = WsgiMethodContext(app,{
+ 'QUERY_STRING': '',
+ 'PATH_INFO': '/some_call',
+ }, 'some-content-type')
+
+ from rpclib.server import ServerBase
+
+ server = ServerBase(app)
+ try:
+ server.get_in_object(ctx)
+ server.get_out_object(ctx)
+ server.get_out_string(ctx)
+ except ValueError:
+ pass
+ else:
+ raise Exception("Must Fail")
+
+ def test_primitive_only(self):
+ class SomeComplexModel(ComplexModel):
+ i = Integer
+ s = String
+
+ class SomeService(ServiceBase):
+ @srpc(SomeComplexModel, _returns=SomeComplexModel)
+ def some_call(scm):
+ return SomeComplexModel(i=5,s='5x')
+
+ app = Application([SomeService], 'tns', Wsdl11(), HttpRpc(), HttpRpc())
+
+ from rpclib.server.wsgi import WsgiMethodContext
+
+ ctx = WsgiMethodContext(app,{
+ 'QUERY_STRING': '',
+ 'PATH_INFO': '/some_call',
+ }, 'some-content-type')
+
+ from rpclib.server import ServerBase
+
+ server = ServerBase(app)
+
+ try:
+ server.get_in_object(ctx)
+ server.get_out_object(ctx)
+ server.get_out_string(ctx)
+ except ValueError:
+ pass
+ else:
+ raise Exception("Must Fail")
+
+
+
+if __name__ == '__main__':
+ unittest.main()
View
51 src/rpclib/test/test_service.py
@@ -17,11 +17,14 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301
#
+#
+# Most of the service tests are performed through the interop tests.
+#
import datetime
import unittest
-import rpclib.interface.wsdl
-import rpclib.protocol.soap
+from rpclib.interface.wsdl import Wsdl11
+from rpclib.protocol.soap import Soap11
from lxml import etree
@@ -124,55 +127,49 @@ class MultipleReturnService(ServiceBase):
def multi(self, s):
return s, 'a', 'b'
-class Test(unittest.TestCase):
- '''Most of the service tests are performed through the interop tests.'''
-
- def _set_up(self):
- self.app = Application([TestService], rpclib.interface.wsdl.Wsdl11,
- rpclib.protocol.soap.Soap11, tns='tns')
+class TestSingle(unittest.TestCase):
+ def setUp(self):
+ self.app = Application([TestService], 'tns', Wsdl11(), Soap11(), Soap11())
self.srv = TestService()
+
self.app.interface.build_interface_document('URL')
- self._wsdl = self.app.interface.get_interface_document()
- self.wsdl = etree.fromstring(self._wsdl)
+ self.wsdl_str = self.app.interface.get_interface_document()
+ self.wsdl_doc = etree.fromstring(self.wsdl_str)
def test_portypes(self):
- self._set_up()
-
- porttype = self.wsdl.find('{http://schemas.xmlsoap.org/wsdl/}portType')
+ porttype = self.wsdl_doc.find('{http://schemas.xmlsoap.org/wsdl/}portType')
self.assertEquals(
len(self.srv.public_methods), len(porttype.getchildren()))
def test_override_param_names(self):
- self._set_up()
+ # FIXME: This test must be rewritten.
for n in ['self', 'import', 'return', 'from']:
- self.assertTrue(n in self._wsdl, '"%s" not in self._wsdl' % n)
+ self.assertTrue(n in self.wsdl_str, '"%s" not in self.wsdl_str' % n)
+
+class TestMultiple(unittest.TestCase):
+ def setUp(self):
+ self.app = Application([MultipleReturnService], 'tns', Wsdl11(), Soap11(), Soap11())
+ self.app.interface.build_interface_document('url')
def test_multiple_return(self):
- app = Application([MultipleReturnService], rpclib.interface.wsdl.Wsdl11,
- rpclib.protocol.soap.Soap11, tns='tns')
- app.interface.build_interface_document('url')
- srv = MultipleReturnService()
- message = srv.public_methods.values()[0].out_message()
+ message_class = MultipleReturnService.public_methods.values()[0].out_message
+ message = message_class()
self.assertEquals(len(message._type_info), 3)
sent_xml = etree.Element('test')
- message.to_parent_element( ('a','b','c'), srv.get_tns(), sent_xml )
+ self.app.out_protocol.to_parent_element(message_class, ('a','b','c'),
+ MultipleReturnService.get_tns(), sent_xml)
sent_xml = sent_xml[0]
print etree.tostring(sent_xml, pretty_print=True)
- response_data = message.from_xml(sent_xml)
+ response_data = self.app.out_protocol.from_element(message_class, sent_xml)
self.assertEquals(len(response_data), 3)
self.assertEqual(response_data[0], 'a')
self.assertEqual(response_data[1], 'b')
self.assertEqual(response_data[2], 'c')
- def test_multiple_ns(self):
- svc = Application([MultipleNamespaceService],rpclib.interface.wsdl.Wsdl11,
- rpclib.protocol.soap.Soap11,tns='tns')
- svc.interface.get_interface_document()
-
if __name__ == '__main__':
unittest.main()
Something went wrong with that request. Please try again.