← Back to branch summary

~certify-web-dev/twisted/certify-trunk

« back to all changes in this revision

Viewing changes to twisted/web2/test/test_xmlrpc.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2007-01-17 14:52:35 UTC
  • mfrom: (1.1.5 upstream) (2.1.2 etch)
  • Revision ID: james.westby@ubuntu.com-20070117145235-btmig6qfmqfen0om
Tags: 2.5.0-0ubuntu1
New upstream version, compatible with python2.5.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/web2/test/test_xmlrpc.py
 
1
# -*- test-case-name: twisted.web.test.test_xmlrpc -*-
 
2
#
 
3
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
 
4
# See LICENSE for details.
 
5
 
 
6
# 
 
7
 
 
8
"""Test XML-RPC support."""
 
9
 
 
10
import xmlrpclib
 
11
 
 
12
from twisted.web2 import xmlrpc
 
13
from twisted.web2.xmlrpc import XMLRPC, addIntrospection
 
14
from twisted.internet import defer
 
15
 
 
16
from twisted.web2.test.test_server import BaseCase
 
17
 
 
18
class TestRuntimeError(RuntimeError):
 
19
    """
 
20
    Fake RuntimeError for testing purposes.
 
21
    """
 
22
 
 
23
class TestValueError(ValueError):
 
24
    """
 
25
    Fake ValueError for testing purposes.
 
26
    """
 
27
 
 
28
class XMLRPCTestResource(XMLRPC):
 
29
    """
 
30
    This is the XML-RPC "server" against which the tests will be run.
 
31
    """
 
32
    FAILURE = 666
 
33
    NOT_FOUND = 23
 
34
    SESSION_EXPIRED = 42
 
35
 
 
36
    addSlash = True # cause it's at the root
 
37
    
 
38
    # the doc string is part of the test
 
39
    def xmlrpc_add(self, request, a, b):
 
40
        """This function add two numbers."""
 
41
        return a + b
 
42
 
 
43
    xmlrpc_add.signature = [['int', 'int', 'int'],
 
44
                            ['double', 'double', 'double']]
 
45
 
 
46
    # the doc string is part of the test
 
47
    def xmlrpc_pair(self, request, string, num):
 
48
        """This function puts the two arguments in an array."""
 
49
        return [string, num]
 
50
 
 
51
    xmlrpc_pair.signature = [['array', 'string', 'int']]
 
52
 
 
53
    # the doc string is part of the test
 
54
    def xmlrpc_defer(self, request, x):
 
55
        """Help for defer."""
 
56
        return defer.succeed(x)
 
57
 
 
58
    def xmlrpc_deferFail(self, request):
 
59
        return defer.fail(TestValueError())
 
60
 
 
61
    # don't add a doc string, it's part of the test
 
62
    def xmlrpc_fail(self, request):
 
63
        raise TestRuntimeError
 
64
 
 
65
    def xmlrpc_fault(self, request):
 
66
        return xmlrpc.Fault(12, "hello")
 
67
 
 
68
    def xmlrpc_deferFault(self, request):
 
69
        return defer.fail(xmlrpc.Fault(17, "hi"))
 
70
 
 
71
    def xmlrpc_complex(self, request):
 
72
        return {"a": ["b", "c", 12, []], "D": "foo"}
 
73
 
 
74
    def xmlrpc_dict(self, request, map, key):
 
75
        return map[key]
 
76
 
 
77
    def getFunction(self, functionPath):
 
78
        try:
 
79
            return XMLRPC.getFunction(self, functionPath)
 
80
        except xmlrpc.NoSuchFunction:
 
81
            if functionPath.startswith("SESSION"):
 
82
                raise xmlrpc.Fault(self.SESSION_EXPIRED, "Session non-existant/expired.")
 
83
            else:
 
84
                raise
 
85
 
 
86
    xmlrpc_dict.help = 'Help for dict.'
 
87
 
 
88
class XMLRPCServerBase(BaseCase):
 
89
    """
 
90
    The parent class of the XML-RPC test classes.
 
91
    """
 
92
    method = 'POST'
 
93
    version = (1, 1)
 
94
 
 
95
    def setUp(self):
 
96
        self.root = XMLRPCTestResource()
 
97
        self.xml = ("<?xml version='1.0'?>\n<methodResponse>\n" +
 
98
            "%s</methodResponse>\n")
 
99
 
 
100
class XMLRPCServerGETTest(XMLRPCServerBase):
 
101
    """
 
102
    Attempt access to the RPC resources as regular HTTP resource.
 
103
    """
 
104
 
 
105
    def setUp(self):
 
106
        super(XMLRPCServerGETTest, self).setUp()
 
107
        self.method = 'GET'
 
108
        self.errorRPC = ('<html><head><title>XML-RPC responder</title>' +
 
109
            '</head><body><h1>XML-RPC responder</h1>POST your XML-RPC ' +
 
110
            'here.</body></html>')
 
111
        self.errorHTTP = ('<html><head><title>404 Not Found</title>' +
 
112
            '</head><body><h1>Not Found</h1>The resource http://host/add ' +
 
113
            'cannot be found.</body></html>')
 
114
 
 
115
    def test_rootGET(self):
 
116
        """
 
117
        Test a simple GET against the XML-RPC server.
 
118
        """
 
119
        return self.assertResponse(
 
120
            (self.root, 'http://host/'),
 
121
            (200, {}, self.errorRPC))
 
122
 
 
123
    def test_childGET(self):
 
124
        """
 
125
        Try to access an XML-RPC method as a regular resource via GET.
 
126
        """
 
127
        return self.assertResponse(
 
128
            (self.root, 'http://host/add'),
 
129
            (404, {}, self.errorHTTP))
 
130
 
 
131
class XMLRPCServerPOSTTest(XMLRPCServerBase):
 
132
    """
 
133
    Tests for standard XML-RPC usage.
 
134
    """
 
135
    def test_RPCMethods(self):
 
136
        """
 
137
        Make RPC calls of the defined methods, checking for the expected 
 
138
        results.
 
139
        """
 
140
        inputOutput = [
 
141
            ("add", (2, 3), 5),
 
142
            ("defer", ("a",), "a"),
 
143
            ("dict", ({"a": 1}, "a"), 1),
 
144
            ("pair", ("a", 1), ["a", 1]),
 
145
            ("complex", (), {"a": ["b", "c", 12, []], "D": "foo"})]
 
146
        dl = []
 
147
        for meth, args, outp in inputOutput:
 
148
            postdata = xmlrpclib.dumps(args, meth)
 
149
            respdata = xmlrpclib.dumps((outp,))
 
150
            reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
 
151
            d = self.assertResponse(reqdata, (200, {}, self.xml % respdata))
 
152
            dl.append(d)
 
153
        return defer.DeferredList(dl, fireOnOneErrback=True)
 
154
 
 
155
    def test_RPCFaults(self):
 
156
        """
 
157
        Ensure that RPC faults are properly processed.
 
158
        """
 
159
        dl = []
 
160
        codeMethod = [
 
161
            (12, "fault", 'hello'),
 
162
            (23, "noSuchMethod", 'function noSuchMethod not found'),
 
163
            (17, "deferFault", 'hi'),
 
164
            (42, "SESSION_TEST", 'Session non-existant/expired.')]
 
165
        for code, meth, fault in codeMethod:
 
166
            postdata = xmlrpclib.dumps((), meth)
 
167
            respdata = xmlrpclib.dumps(xmlrpc.Fault(code, fault))
 
168
            reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
 
169
            d = self.assertResponse(reqdata, (200, {}, respdata))
 
170
            dl.append(d)
 
171
        d = defer.DeferredList(dl, fireOnOneErrback=True)
 
172
        return d
 
173
 
 
174
    def test_RPCFailures(self):
 
175
        """
 
176
        Ensure that failures behave as expected.
 
177
        """
 
178
        dl = []
 
179
        codeMethod = [
 
180
            (666, "fail"),
 
181
            (666, "deferFail")]
 
182
        for code, meth in codeMethod:
 
183
            postdata = xmlrpclib.dumps((), meth)
 
184
            respdata = xmlrpclib.dumps(xmlrpc.Fault(code, 'error'))
 
185
            reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
 
186
            d = self.assertResponse(reqdata, (200, {}, respdata))
 
187
            d.addCallback(self.flushLoggedErrors, TestRuntimeError, TestValueError)
 
188
            dl.append(d)
 
189
        d = defer.DeferredList(dl, fireOnOneErrback=True)
 
190
        return d
 
191
 
 
192
class XMLRPCTestIntrospection(XMLRPCServerBase):
 
193
 
 
194
    def setUp(self):
 
195
        """
 
196
        Introspection requires additional setup, most importantly, adding
 
197
        introspection to the root object.
 
198
        """
 
199
        super(XMLRPCTestIntrospection, self).setUp()
 
200
        addIntrospection(self.root)
 
201
        self.methodList = ['add', 'complex', 'defer', 'deferFail',
 
202
            'deferFault', 'dict', 'fail', 'fault', 'pair',
 
203
            'system.listMethods', 'system.methodHelp', 'system.methodSignature']
 
204
 
 
205
    def test_listMethods(self):
 
206
        """
 
207
        Check that the introspection method "listMethods" returns all the
 
208
        methods we defined in the XML-RPC server.
 
209
        """
 
210
        def cbMethods(meths):
 
211
            meths.sort()
 
212
            self.failUnlessEqual(
 
213
                meths,
 
214
                )
 
215
        postdata = xmlrpclib.dumps((), 'system.listMethods')
 
216
        respdata = xmlrpclib.dumps((self.methodList,))
 
217
        reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
 
218
        return self.assertResponse(reqdata, (200, {}, self.xml % respdata))
 
219
 
 
220
    def test_methodHelp(self):
 
221
        """
 
222
        Check the RPC methods for docstrings or .help attributes.
 
223
        """
 
224
        inputOutput = [
 
225
            ("defer", "Help for defer."),
 
226
            ("fail", ""),
 
227
            ("dict", "Help for dict.")]
 
228
 
 
229
        dl = []
 
230
        for meth, outp in inputOutput:
 
231
            postdata = xmlrpclib.dumps((meth,), 'system.methodHelp')
 
232
            respdata = xmlrpclib.dumps((outp,))
 
233
            reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
 
234
            d = self.assertResponse(reqdata, (200, {}, self.xml % respdata))
 
235
            dl.append(d)
 
236
        return defer.DeferredList(dl, fireOnOneErrback=True)
 
237
 
 
238
    def test_methodSignature(self):
 
239
        """
 
240
        Check that the RPC methods whose signatures have been set via the
 
241
        .signature attribute (on the method) are returned as expected.
 
242
        """
 
243
        inputOutput = [
 
244
            ("defer", ""),
 
245
            ("add", [['int', 'int', 'int'],
 
246
                     ['double', 'double', 'double']]),
 
247
            ("pair", [['array', 'string', 'int']])]
 
248
 
 
249
        dl = []
 
250
        for meth, outp in inputOutput:
 
251
            postdata = xmlrpclib.dumps((meth,), 'system.methodSignature')
 
252
            respdata = xmlrpclib.dumps((outp,))
 
253
            reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
 
254
            d = self.assertResponse(reqdata, (200, {}, self.xml % respdata))
 
255
            dl.append(d)
 
256
        return defer.DeferredList(dl, fireOnOneErrback=True)
 
257
 
 
258