tests/python/testxxmlrpc.py

   1 #!/usr/bin/env python2.3
   2 from __future__ import generators
   3 import unittest
   4 import sys, StringIO
   5 import os, time, xmlrpclib
   6 from os.path import dirname, abspath, join
   7
   8 rox_lib = dirname(dirname(dirname(abspath(sys.argv[0]))))
   9 sys.path.insert(0, join(rox_lib, 'python'))
  10
  11 from rox import xxmlrpc, g
  12
  13 class TestObject:
  14         allowed_methods = ['echo', 'none']
  15
  16         def echo(self, msg):
  17                 #print "Got", msg
  18                 return "Echo: " + msg
  19
  20         def none(self):
  21                 return
  22
  23 class TestXXMLRPC(unittest.TestCase):
  24         def setUp(self):
  25                 service = xxmlrpc.XXMLRPCServer('rox_test_service')
  26                 service.add_object('/foo', TestObject())
  27                 self.proxy = xxmlrpc.XXMLProxy('rox_test_service')
  28
  29         def testEcho(self):
  30                 obj = self.proxy.get_object('/foo')
  31                 call = obj.echo('Hello World')
  32                 self.assertEquals("Echo: Hello World", call.get_response())
  33
  34         def testFault(self):
  35                 obj = self.proxy.get_object('/foo')
  36                 call = obj.echo(0)
  37                 try:
  38                         call.get_response()
  39                         assert false
  40                 except xmlrpclib.Fault, ex:
  41                         self.assertEquals('TypeError', ex.faultCode)
  42                         assert ex.faultString.find('cannot concatenate') >= 0
  43
  44         def testAsync(self):
  45                 obj = self.proxy.get_object('/foo')
  46                 call1 = obj.echo('Hello')
  47                 call2 = obj.echo('World')
  48                 self.assertEquals("Echo: World", call2.get_response())
  49                 self.assertEquals("Echo: Hello", call1.get_response())
  50
  51         def testBadObject(self):
  52                 obj = self.proxy.get_object('/food')
  53                 call = obj.echo(0)
  54                 try:
  55                         call.get_response()
  56                         assert false
  57                 except xmlrpclib.Fault, ex:
  58                         self.assertEquals('UnknownObject', ex.faultCode)
  59
  60         def testBadMethod(self):
  61                 obj = self.proxy.get_object('/foo')
  62                 call = obj.write("Hi")
  63                 try:
  64                         call.get_response()
  65                         assert false
  66                 except xmlrpclib.Fault, ex:
  67                         self.assertEquals('NoSuchMethod', ex.faultCode)
  68
  69         def testReturnNone(self):
  70                 obj = self.proxy.get_object('/foo')
  71                 call = obj.none()
  72                 self.assertEquals(True, call.get_response())
  73
  74         def testNoReturn(self):
  75                 obj = self.proxy.get_object('/foo')
  76                 call = obj.none()
  77                 olderr = sys.stderr
  78                 sys.stderr = StringIO.StringIO()
  79                 try:
  80                         del call
  81                         err = sys.stderr.getvalue()
  82                         assert err.index("ClientCall object destroyed") == 0
  83                         sys.stderr = StringIO.StringIO()
  84                         # Wait for proxy to try to read window
  85                         while not sys.stderr.getvalue():
  86                                 g.main_iteration()
  87                         err = sys.stderr.getvalue()
  88                         assert err.index("No '_XXMLRPC_MESSAGE' property") == 0
  89                 finally:
  90                         sys.stderr = olderr
  91
  92 suite = unittest.makeSuite(TestXXMLRPC)
  93 if __name__ == '__main__':
  94         sys.argv.append('-v')
  95         unittest.main()