monitor/cherrypy/test/test_xmlrpc.py

   1 import sys
   2 from xmlrpclib import DateTime, Fault, ServerProxy, SafeTransport
   3
   4 class HTTPSTransport(SafeTransport):
   5     """Subclass of SafeTransport to fix sock.recv errors (by using file)."""
   6
   7     def request(self, host, handler, request_body, verbose=0):
   8         # issue XML-RPC request
   9         h = self.make_connection(host)
  10         if verbose:
  11             h.set_debuglevel(1)
  12
  13         self.send_request(h, handler, request_body)
  14         self.send_host(h, host)
  15         self.send_user_agent(h)
  16         self.send_content(h, request_body)
  17
  18         errcode, errmsg, headers = h.getreply()
  19         if errcode != 200:
  20             raise xmlrpclib.ProtocolError(host + handler, errcode, errmsg,
  21                                           headers)
  22
  23         self.verbose = verbose
  24
  25         # Here's where we differ from the superclass. It says:
  26         # try:
  27         #     sock = h._conn.sock
  28         # except AttributeError:
  29         #     sock = None
  30         # return self._parse_response(h.getfile(), sock)
  31
  32         return self.parse_response(h.getfile())
  33
  34 import cherrypy
  35
  36
  37 def setup_server():
  38     from cherrypy import _cptools
  39
  40     class Root:
  41         def index(self):
  42             return "I'm a standard index!"
  43         index.exposed = True
  44
  45
  46     class XmlRpc(_cptools.XMLRPCController):
  47
  48         def foo(self):
  49             return "Hello world!"
  50         foo.exposed = True
  51
  52         def return_single_item_list(self):
  53             return [42]
  54         return_single_item_list.exposed = True
  55
  56         def return_string(self):
  57             return "here is a string"
  58         return_string.exposed = True
  59
  60         def return_tuple(self):
  61             return ('here', 'is', 1, 'tuple')
  62         return_tuple.exposed = True
  63
  64         def return_dict(self):
  65             return dict(a=1, b=2, c=3)
  66         return_dict.exposed = True
  67
  68         def return_composite(self):
  69             return dict(a=1,z=26), 'hi', ['welcome', 'friend']
  70         return_composite.exposed = True
  71
  72         def return_int(self):
  73             return 42
  74         return_int.exposed = True
  75
  76         def return_float(self):
  77             return 3.14
  78         return_float.exposed = True
  79
  80         def return_datetime(self):
  81             return DateTime((2003, 10, 7, 8, 1, 0, 1, 280, -1))
  82         return_datetime.exposed = True
  83
  84         def return_boolean(self):
  85             return True
  86         return_boolean.exposed = True
  87
  88         def test_argument_passing(self, num):
  89             return num * 2
  90         test_argument_passing.exposed = True
  91
  92         def test_returning_Fault(self):
  93             return Fault(1, "custom Fault response")
  94         test_returning_Fault.exposed = True
  95
  96     root = Root()
  97     root.xmlrpc = XmlRpc()
  98     cherrypy.tree.mount(root, config={'/': {
  99         'request.dispatch': cherrypy.dispatch.XMLRPCDispatcher(),
 100         'tools.xmlrpc.allow_none': 0,
 101         }})
 102
 103
 104 from cherrypy.test import helper
 105
 106 class XmlRpcTest(helper.CPWebCase):
 107     setup_server = staticmethod(setup_server)
 108     def testXmlRpc(self):
 109
 110         scheme = "http"
 111         try:
 112             scheme = self.harness.scheme
 113         except AttributeError:
 114             pass
 115
 116         if scheme == "https":
 117             url = 'https://%s:%s/xmlrpc/' % (self.interface(), self.PORT)
 118             proxy = ServerProxy(url, transport=HTTPSTransport())
 119         else:
 120             url = 'http://%s:%s/xmlrpc/' % (self.interface(), self.PORT)
 121             proxy = ServerProxy(url)
 122
 123         # begin the tests ...
 124         self.getPage("/xmlrpc/foo")
 125         self.assertBody("Hello world!")
 126
 127         self.assertEqual(proxy.return_single_item_list(), [42])
 128         self.assertNotEqual(proxy.return_single_item_list(), 'one bazillion')
 129         self.assertEqual(proxy.return_string(), "here is a string")
 130         self.assertEqual(proxy.return_tuple(), list(('here', 'is', 1, 'tuple')))
 131         self.assertEqual(proxy.return_dict(), {'a': 1, 'c': 3, 'b': 2})
 132         self.assertEqual(proxy.return_composite(),
 133                          [{'a': 1, 'z': 26}, 'hi', ['welcome', 'friend']])
 134         self.assertEqual(proxy.return_int(), 42)
 135         self.assertEqual(proxy.return_float(), 3.14)
 136         self.assertEqual(proxy.return_datetime(),
 137                          DateTime((2003, 10, 7, 8, 1, 0, 1, 280, -1)))
 138         self.assertEqual(proxy.return_boolean(), True)
 139         self.assertEqual(proxy.test_argument_passing(22), 22 * 2)
 140
 141         # Test an error in the page handler (should raise an xmlrpclib.Fault)
 142         try:
 143             proxy.test_argument_passing({})
 144         except Exception:
 145             x = sys.exc_info()[1]
 146             self.assertEqual(x.__class__, Fault)
 147             self.assertEqual(x.faultString, ("unsupported operand type(s) "
 148                                              "for *: 'dict' and 'int'"))
 149         else:
 150             self.fail("Expected xmlrpclib.Fault")
 151
 152         # http://www.cherrypy.org/ticket/533
 153         # if a method is not found, an xmlrpclib.Fault should be raised
 154         try:
 155             proxy.non_method()
 156         except Exception:
 157             x = sys.exc_info()[1]
 158             self.assertEqual(x.__class__, Fault)
 159             self.assertEqual(x.faultString, 'method "non_method" is not supported')
 160         else:
 161             self.fail("Expected xmlrpclib.Fault")
 162
 163         # Test returning a Fault from the page handler.
 164         try:
 165             proxy.test_returning_Fault()
 166         except Exception:
 167             x = sys.exc_info()[1]
 168             self.assertEqual(x.__class__, Fault)
 169             self.assertEqual(x.faultString, ("custom Fault response"))
 170         else:
 171             self.fail("Expected xmlrpclib.Fault")
 172