# -*- test-case-name: twisted.web.test.test_xmlrpc -*-
#
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
# See LICENSE for details.
#
"""Test XML-RPC support."""
import xmlrpclib
from twisted.web2 import xmlrpc
from twisted.web2.xmlrpc import XMLRPC, addIntrospection
from twisted.internet import defer
from twisted.web2.test.test_server import BaseCase
class TestRuntimeError(RuntimeError):
"""
Fake RuntimeError for testing purposes.
"""
class TestValueError(ValueError):
"""
Fake ValueError for testing purposes.
"""
class XMLRPCTestResource(XMLRPC):
"""
This is the XML-RPC "server" against which the tests will be run.
"""
FAILURE = 666
NOT_FOUND = 23
SESSION_EXPIRED = 42
addSlash = True # cause it's at the root
# the doc string is part of the test
def xmlrpc_add(self, request, a, b):
"""This function add two numbers."""
return a + b
xmlrpc_add.signature = [['int', 'int', 'int'],
['double', 'double', 'double']]
# the doc string is part of the test
def xmlrpc_pair(self, request, string, num):
"""This function puts the two arguments in an array."""
return [string, num]
xmlrpc_pair.signature = [['array', 'string', 'int']]
# the doc string is part of the test
def xmlrpc_defer(self, request, x):
"""Help for defer."""
return defer.succeed(x)
def xmlrpc_deferFail(self, request):
return defer.fail(TestValueError())
# don't add a doc string, it's part of the test
def xmlrpc_fail(self, request):
raise TestRuntimeError
def xmlrpc_fault(self, request):
return xmlrpc.Fault(12, "hello")
def xmlrpc_deferFault(self, request):
return defer.fail(xmlrpc.Fault(17, "hi"))
def xmlrpc_complex(self, request):
return {"a": ["b", "c", 12, []], "D": "foo"}
def xmlrpc_dict(self, request, map, key):
return map[key]
def getFunction(self, functionPath):
try:
return XMLRPC.getFunction(self, functionPath)
except xmlrpc.NoSuchFunction:
if functionPath.startswith("SESSION"):
raise xmlrpc.Fault(self.SESSION_EXPIRED, "Session non-existant/expired.")
else:
raise
xmlrpc_dict.help = 'Help for dict.'
class XMLRPCServerBase(BaseCase):
"""
The parent class of the XML-RPC test classes.
"""
method = 'POST'
version = (1, 1)
def setUp(self):
self.root = XMLRPCTestResource()
self.xml = ("\n\n" +
"%s\n")
class XMLRPCServerGETTest(XMLRPCServerBase):
"""
Attempt access to the RPC resources as regular HTTP resource.
"""
def setUp(self):
super(XMLRPCServerGETTest, self).setUp()
self.method = 'GET'
self.errorRPC = ('XML-RPC responder' +
'XML-RPC responder
POST your XML-RPC ' +
'here.')
self.errorHTTP = ('404 Not Found' +
'Not Found
The resource http://host/add ' +
'cannot be found.')
def test_rootGET(self):
"""
Test a simple GET against the XML-RPC server.
"""
return self.assertResponse(
(self.root, 'http://host/'),
(200, {}, self.errorRPC))
def test_childGET(self):
"""
Try to access an XML-RPC method as a regular resource via GET.
"""
return self.assertResponse(
(self.root, 'http://host/add'),
(404, {}, self.errorHTTP))
class XMLRPCServerPOSTTest(XMLRPCServerBase):
"""
Tests for standard XML-RPC usage.
"""
def test_RPCMethods(self):
"""
Make RPC calls of the defined methods, checking for the expected
results.
"""
inputOutput = [
("add", (2, 3), 5),
("defer", ("a",), "a"),
("dict", ({"a": 1}, "a"), 1),
("pair", ("a", 1), ["a", 1]),
("complex", (), {"a": ["b", "c", 12, []], "D": "foo"})]
dl = []
for meth, args, outp in inputOutput:
postdata = xmlrpclib.dumps(args, meth)
respdata = xmlrpclib.dumps((outp,))
reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
d = self.assertResponse(reqdata, (200, {}, self.xml % respdata))
dl.append(d)
return defer.DeferredList(dl, fireOnOneErrback=True)
def test_RPCFaults(self):
"""
Ensure that RPC faults are properly processed.
"""
dl = []
codeMethod = [
(12, "fault", 'hello'),
(23, "noSuchMethod", 'function noSuchMethod not found'),
(17, "deferFault", 'hi'),
(42, "SESSION_TEST", 'Session non-existant/expired.')]
for code, meth, fault in codeMethod:
postdata = xmlrpclib.dumps((), meth)
respdata = xmlrpclib.dumps(xmlrpc.Fault(code, fault))
reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
d = self.assertResponse(reqdata, (200, {}, respdata))
dl.append(d)
d = defer.DeferredList(dl, fireOnOneErrback=True)
return d
def test_RPCFailures(self):
"""
Ensure that failures behave as expected.
"""
dl = []
codeMethod = [
(666, "fail"),
(666, "deferFail")]
for code, meth in codeMethod:
postdata = xmlrpclib.dumps((), meth)
respdata = xmlrpclib.dumps(xmlrpc.Fault(code, 'error'))
reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
d = self.assertResponse(reqdata, (200, {}, respdata))
d.addCallback(self.flushLoggedErrors, TestRuntimeError, TestValueError)
dl.append(d)
d = defer.DeferredList(dl, fireOnOneErrback=True)
return d
class XMLRPCTestIntrospection(XMLRPCServerBase):
def setUp(self):
"""
Introspection requires additional setup, most importantly, adding
introspection to the root object.
"""
super(XMLRPCTestIntrospection, self).setUp()
addIntrospection(self.root)
self.methodList = ['add', 'complex', 'defer', 'deferFail',
'deferFault', 'dict', 'fail', 'fault', 'pair',
'system.listMethods', 'system.methodHelp', 'system.methodSignature']
def test_listMethods(self):
"""
Check that the introspection method "listMethods" returns all the
methods we defined in the XML-RPC server.
"""
def cbMethods(meths):
meths.sort()
self.failUnlessEqual(
meths,
)
postdata = xmlrpclib.dumps((), 'system.listMethods')
respdata = xmlrpclib.dumps((self.methodList,))
reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
return self.assertResponse(reqdata, (200, {}, self.xml % respdata))
def test_methodHelp(self):
"""
Check the RPC methods for docstrings or .help attributes.
"""
inputOutput = [
("defer", "Help for defer."),
("fail", ""),
("dict", "Help for dict.")]
dl = []
for meth, outp in inputOutput:
postdata = xmlrpclib.dumps((meth,), 'system.methodHelp')
respdata = xmlrpclib.dumps((outp,))
reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
d = self.assertResponse(reqdata, (200, {}, self.xml % respdata))
dl.append(d)
return defer.DeferredList(dl, fireOnOneErrback=True)
def test_methodSignature(self):
"""
Check that the RPC methods whose signatures have been set via the
.signature attribute (on the method) are returned as expected.
"""
inputOutput = [
("defer", ""),
("add", [['int', 'int', 'int'],
['double', 'double', 'double']]),
("pair", [['array', 'string', 'int']])]
dl = []
for meth, outp in inputOutput:
postdata = xmlrpclib.dumps((meth,), 'system.methodSignature')
respdata = xmlrpclib.dumps((outp,))
reqdata = (self.root, 'http://host/', {}, None, None, '', postdata)
d = self.assertResponse(reqdata, (200, {}, self.xml % respdata))
dl.append(d)
return defer.DeferredList(dl, fireOnOneErrback=True)