Lib/test/test_xmlrpc.py

   1 import base64
   2 import datetime
   3 import sys
   4 import time
   5 import unittest
   6 import xmlrpclib
   7 import SimpleXMLRPCServer
   8 import threading
   9 import mimetools
  10 import httplib
  11 import socket
  12 import StringIO
  13 import os
  14 import re
  15 from test import test_support
  16
  17 try:
  18     unicode
  19 except NameError:
  20     have_unicode = False
  21 else:
  22     have_unicode = True
  23
  24 alist = [{'astring': 'foo@bar.baz.spam',
  25           'afloat': 7283.43,
  26           'anint': 2**20,
  27           'ashortlong': 2L,
  28           'anotherlist': ['.zyx.41'],
  29           'abase64': xmlrpclib.Binary("my dog has fleas"),
  30           'boolean': xmlrpclib.False,
  31           'unicode': u'\u4000\u6000\u8000',
  32           u'ukey\u4000': 'regular value',
  33           'datetime1': xmlrpclib.DateTime('20050210T11:41:23'),
  34           'datetime2': xmlrpclib.DateTime(
  35                         (2005, 02, 10, 11, 41, 23, 0, 1, -1)),
  36           'datetime3': xmlrpclib.DateTime(
  37                         datetime.datetime(2005, 02, 10, 11, 41, 23)),
  38           }]
  39
  40 class XMLRPCTestCase(unittest.TestCase):
  41
  42     def test_dump_load(self):
  43         self.assertEquals(alist,
  44                           xmlrpclib.loads(xmlrpclib.dumps((alist,)))[0][0])
  45
  46     def test_dump_bare_datetime(self):
  47         # This checks that an unwrapped datetime.date object can be handled
  48         # by the marshalling code.  This can't be done via test_dump_load()
  49         # since with use_datetime set to 1 the unmarshaller would create
  50         # datetime objects for the 'datetime[123]' keys as well
  51         dt = datetime.datetime(2005, 02, 10, 11, 41, 23)
  52         s = xmlrpclib.dumps((dt,))
  53         (newdt,), m = xmlrpclib.loads(s, use_datetime=1)
  54         self.assertEquals(newdt, dt)
  55         self.assertEquals(m, None)
  56
  57         (newdt,), m = xmlrpclib.loads(s, use_datetime=0)
  58         self.assertEquals(newdt, xmlrpclib.DateTime('20050210T11:41:23'))
  59
  60     def test_datetime_before_1900(self):
  61         # same as before but with a date before 1900
  62         dt = datetime.datetime(1, 02, 10, 11, 41, 23)
  63         s = xmlrpclib.dumps((dt,))
  64         (newdt,), m = xmlrpclib.loads(s, use_datetime=1)
  65         self.assertEquals(newdt, dt)
  66         self.assertEquals(m, None)
  67
  68         (newdt,), m = xmlrpclib.loads(s, use_datetime=0)
  69         self.assertEquals(newdt, xmlrpclib.DateTime('00010210T11:41:23'))
  70
  71     def test_cmp_datetime_DateTime(self):
  72         now = datetime.datetime.now()
  73         dt = xmlrpclib.DateTime(now.timetuple())
  74         self.assertTrue(dt == now)
  75         self.assertTrue(now == dt)
  76         then = now + datetime.timedelta(seconds=4)
  77         self.assertTrue(then >= dt)
  78         self.assertTrue(dt < then)
  79
  80     def test_bug_1164912 (self):
  81         d = xmlrpclib.DateTime()
  82         ((new_d,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((d,),
  83                                             methodresponse=True))
  84         self.assertTrue(isinstance(new_d.value, str))
  85
  86         # Check that the output of dumps() is still an 8-bit string
  87         s = xmlrpclib.dumps((new_d,), methodresponse=True)
  88         self.assertTrue(isinstance(s, str))
  89
  90     def test_newstyle_class(self):
  91         class T(object):
  92             pass
  93         t = T()
  94         t.x = 100
  95         t.y = "Hello"
  96         ((t2,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((t,)))
  97         self.assertEquals(t2, t.__dict__)
  98
  99     def test_dump_big_long(self):
 100         self.assertRaises(OverflowError, xmlrpclib.dumps, (2L**99,))
 101
 102     def test_dump_bad_dict(self):
 103         self.assertRaises(TypeError, xmlrpclib.dumps, ({(1,2,3): 1},))
 104
 105     def test_dump_recursive_seq(self):
 106         l = [1,2,3]
 107         t = [3,4,5,l]
 108         l.append(t)
 109         self.assertRaises(TypeError, xmlrpclib.dumps, (l,))
 110
 111     def test_dump_recursive_dict(self):
 112         d = {'1':1, '2':1}
 113         t = {'3':3, 'd':d}
 114         d['t'] = t
 115         self.assertRaises(TypeError, xmlrpclib.dumps, (d,))
 116
 117     def test_dump_big_int(self):
 118         if sys.maxint > 2L**31-1:
 119             self.assertRaises(OverflowError, xmlrpclib.dumps,
 120                               (int(2L**34),))
 121
 122         xmlrpclib.dumps((xmlrpclib.MAXINT, xmlrpclib.MININT))
 123         self.assertRaises(OverflowError, xmlrpclib.dumps, (xmlrpclib.MAXINT+1,))
 124         self.assertRaises(OverflowError, xmlrpclib.dumps, (xmlrpclib.MININT-1,))
 125
 126         def dummy_write(s):
 127             pass
 128
 129         m = xmlrpclib.Marshaller()
 130         m.dump_int(xmlrpclib.MAXINT, dummy_write)
 131         m.dump_int(xmlrpclib.MININT, dummy_write)
 132         self.assertRaises(OverflowError, m.dump_int, xmlrpclib.MAXINT+1, dummy_write)
 133         self.assertRaises(OverflowError, m.dump_int, xmlrpclib.MININT-1, dummy_write)
 134
 135
 136     def test_dump_none(self):
 137         value = alist + [None]
 138         arg1 = (alist + [None],)
 139         strg = xmlrpclib.dumps(arg1, allow_none=True)
 140         self.assertEquals(value,
 141                           xmlrpclib.loads(strg)[0][0])
 142         self.assertRaises(TypeError, xmlrpclib.dumps, (arg1,))
 143
 144     def test_default_encoding_issues(self):
 145         # SF bug #1115989: wrong decoding in '_stringify'
 146         utf8 = """<?xml version='1.0' encoding='iso-8859-1'?>
 147                   <params>
 148                     <param><value>
 149                       <string>abc \x95</string>
 150                       </value></param>
 151                     <param><value>
 152                       <struct>
 153                         <member>
 154                           <name>def \x96</name>
 155                           <value><string>ghi \x97</string></value>
 156                           </member>
 157                         </struct>
 158                       </value></param>
 159                   </params>
 160                   """
 161
 162         # sys.setdefaultencoding() normally doesn't exist after site.py is
 163         # loaded.  Import a temporary fresh copy to get access to it
 164         # but then restore the original copy to avoid messing with
 165         # other potentially modified sys module attributes
 166         old_encoding = sys.getdefaultencoding()
 167         with test_support.CleanImport('sys'):
 168             import sys as temp_sys
 169             temp_sys.setdefaultencoding("iso-8859-1")
 170             try:
 171                 (s, d), m = xmlrpclib.loads(utf8)
 172             finally:
 173                 temp_sys.setdefaultencoding(old_encoding)
 174
 175         items = d.items()
 176         if have_unicode:
 177             self.assertEquals(s, u"abc \x95")
 178             self.assertTrue(isinstance(s, unicode))
 179             self.assertEquals(items, [(u"def \x96", u"ghi \x97")])
 180             self.assertTrue(isinstance(items[0][0], unicode))
 181             self.assertTrue(isinstance(items[0][1], unicode))
 182         else:
 183             self.assertEquals(s, "abc \xc2\x95")
 184             self.assertEquals(items, [("def \xc2\x96", "ghi \xc2\x97")])
 185
 186
 187 class HelperTestCase(unittest.TestCase):
 188     def test_escape(self):
 189         self.assertEqual(xmlrpclib.escape("a&b"), "a&amp;b")
 190         self.assertEqual(xmlrpclib.escape("a<b"), "a&lt;b")
 191         self.assertEqual(xmlrpclib.escape("a>b"), "a&gt;b")
 192
 193 class FaultTestCase(unittest.TestCase):
 194     def test_repr(self):
 195         f = xmlrpclib.Fault(42, 'Test Fault')
 196         self.assertEqual(repr(f), "<Fault 42: 'Test Fault'>")
 197         self.assertEqual(repr(f), str(f))
 198
 199     def test_dump_fault(self):
 200         f = xmlrpclib.Fault(42, 'Test Fault')
 201         s = xmlrpclib.dumps((f,))
 202         (newf,), m = xmlrpclib.loads(s)
 203         self.assertEquals(newf, {'faultCode': 42, 'faultString': 'Test Fault'})
 204         self.assertEquals(m, None)
 205
 206         s = xmlrpclib.Marshaller().dumps(f)
 207         self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, s)
 208
 209
 210 class DateTimeTestCase(unittest.TestCase):
 211     def test_default(self):
 212         t = xmlrpclib.DateTime()
 213
 214     def test_time(self):
 215         d = 1181399930.036952
 216         t = xmlrpclib.DateTime(d)
 217         self.assertEqual(str(t), time.strftime("%Y%m%dT%H:%M:%S", time.localtime(d)))
 218
 219     def test_time_tuple(self):
 220         d = (2007,6,9,10,38,50,5,160,0)
 221         t = xmlrpclib.DateTime(d)
 222         self.assertEqual(str(t), '20070609T10:38:50')
 223
 224     def test_time_struct(self):
 225         d = time.localtime(1181399930.036952)
 226         t = xmlrpclib.DateTime(d)
 227         self.assertEqual(str(t),  time.strftime("%Y%m%dT%H:%M:%S", d))
 228
 229     def test_datetime_datetime(self):
 230         d = datetime.datetime(2007,1,2,3,4,5)
 231         t = xmlrpclib.DateTime(d)
 232         self.assertEqual(str(t), '20070102T03:04:05')
 233
 234     def test_repr(self):
 235         d = datetime.datetime(2007,1,2,3,4,5)
 236         t = xmlrpclib.DateTime(d)
 237         val ="<DateTime '20070102T03:04:05' at %x>" % id(t)
 238         self.assertEqual(repr(t), val)
 239
 240     def test_decode(self):
 241         d = ' 20070908T07:11:13  '
 242         t1 = xmlrpclib.DateTime()
 243         t1.decode(d)
 244         tref = xmlrpclib.DateTime(datetime.datetime(2007,9,8,7,11,13))
 245         self.assertEqual(t1, tref)
 246
 247         t2 = xmlrpclib._datetime(d)
 248         self.assertEqual(t1, tref)
 249
 250 class BinaryTestCase(unittest.TestCase):
 251     def test_default(self):
 252         t = xmlrpclib.Binary()
 253         self.assertEqual(str(t), '')
 254
 255     def test_string(self):
 256         d = '\x01\x02\x03abc123\xff\xfe'
 257         t = xmlrpclib.Binary(d)
 258         self.assertEqual(str(t), d)
 259
 260     def test_decode(self):
 261         d = '\x01\x02\x03abc123\xff\xfe'
 262         de = base64.encodestring(d)
 263         t1 = xmlrpclib.Binary()
 264         t1.decode(de)
 265         self.assertEqual(str(t1), d)
 266
 267         t2 = xmlrpclib._binary(de)
 268         self.assertEqual(str(t2), d)
 269
 270
 271 ADDR = PORT = URL = None
 272
 273 # The evt is set twice.  First when the server is ready to serve.
 274 # Second when the server has been shutdown.  The user must clear
 275 # the event after it has been set the first time to catch the second set.
 276 def http_server(evt, numrequests, requestHandler=None):
 277     class TestInstanceClass:
 278         def div(self, x, y):
 279             return x // y
 280
 281         def _methodHelp(self, name):
 282             if name == 'div':
 283                 return 'This is the div function'
 284
 285     def my_function():
 286         '''This is my function'''
 287         return True
 288
 289     class MyXMLRPCServer(SimpleXMLRPCServer.SimpleXMLRPCServer):
 290         def get_request(self):
 291             # Ensure the socket is always non-blocking.  On Linux, socket
 292             # attributes are not inherited like they are on *BSD and Windows.
 293             s, port = self.socket.accept()
 294             s.setblocking(True)
 295             return s, port
 296
 297     if not requestHandler:
 298         requestHandler = SimpleXMLRPCServer.SimpleXMLRPCRequestHandler
 299     serv = MyXMLRPCServer(("localhost", 0), requestHandler,
 300                           logRequests=False, bind_and_activate=False)
 301     try:
 302         serv.socket.settimeout(3)
 303         serv.server_bind()
 304         global ADDR, PORT, URL
 305         ADDR, PORT = serv.socket.getsockname()
 306         #connect to IP address directly.  This avoids socket.create_connection()
 307         #trying to connect to to "localhost" using all address families, which
 308         #causes slowdown e.g. on vista which supports AF_INET6.  The server listens
 309         #on AF_INET only.
 310         URL = "http://%s:%d"%(ADDR, PORT)
 311         serv.server_activate()
 312         serv.register_introspection_functions()
 313         serv.register_multicall_functions()
 314         serv.register_function(pow)
 315         serv.register_function(lambda x,y: x+y, 'add')
 316         serv.register_function(my_function)
 317         serv.register_instance(TestInstanceClass())
 318         evt.set()
 319
 320         # handle up to 'numrequests' requests
 321         while numrequests > 0:
 322             serv.handle_request()
 323             numrequests -= 1
 324
 325     except socket.timeout:
 326         pass
 327     finally:
 328         serv.socket.close()
 329         PORT = None
 330         evt.set()
 331
 332 def http_multi_server(evt, numrequests, requestHandler=None):
 333     class TestInstanceClass:
 334         def div(self, x, y):
 335             return x // y
 336
 337         def _methodHelp(self, name):
 338             if name == 'div':
 339                 return 'This is the div function'
 340
 341     def my_function():
 342         '''This is my function'''
 343         return True
 344
 345     class MyXMLRPCServer(SimpleXMLRPCServer.MultiPathXMLRPCServer):
 346         def get_request(self):
 347             # Ensure the socket is always non-blocking.  On Linux, socket
 348             # attributes are not inherited like they are on *BSD and Windows.
 349             s, port = self.socket.accept()
 350             s.setblocking(True)
 351             return s, port
 352
 353     if not requestHandler:
 354         requestHandler = SimpleXMLRPCServer.SimpleXMLRPCRequestHandler
 355     class MyRequestHandler(requestHandler):
 356         rpc_paths = []
 357
 358     serv = MyXMLRPCServer(("localhost", 0), MyRequestHandler,
 359                           logRequests=False, bind_and_activate=False)
 360     serv.socket.settimeout(3)
 361     serv.server_bind()
 362     try:
 363         global ADDR, PORT, URL
 364         ADDR, PORT = serv.socket.getsockname()
 365         #connect to IP address directly.  This avoids socket.create_connection()
 366         #trying to connect to to "localhost" using all address families, which
 367         #causes slowdown e.g. on vista which supports AF_INET6.  The server listens
 368         #on AF_INET only.
 369         URL = "http://%s:%d"%(ADDR, PORT)
 370         serv.server_activate()
 371         paths = ["/foo", "/foo/bar"]
 372         for path in paths:
 373             d = serv.add_dispatcher(path, SimpleXMLRPCServer.SimpleXMLRPCDispatcher())
 374             d.register_introspection_functions()
 375             d.register_multicall_functions()
 376         serv.get_dispatcher(paths[0]).register_function(pow)
 377         serv.get_dispatcher(paths[1]).register_function(lambda x,y: x+y, 'add')
 378         evt.set()
 379
 380         # handle up to 'numrequests' requests
 381         while numrequests > 0:
 382             serv.handle_request()
 383             numrequests -= 1
 384
 385     except socket.timeout:
 386         pass
 387     finally:
 388         serv.socket.close()
 389         PORT = None
 390         evt.set()
 391
 392 # This function prevents errors like:
 393 #    <ProtocolError for localhost:57527/RPC2: 500 Internal Server Error>
 394 def is_unavailable_exception(e):
 395     '''Returns True if the given ProtocolError is the product of a server-side
 396        exception caused by the 'temporarily unavailable' response sometimes
 397        given by operations on non-blocking sockets.'''
 398
 399     # sometimes we get a -1 error code and/or empty headers
 400     try:
 401         if e.errcode == -1 or e.headers is None:
 402             return True
 403         exc_mess = e.headers.get('X-exception')
 404     except AttributeError:
 405         # Ignore socket.errors here.
 406         exc_mess = str(e)
 407
 408     if exc_mess and 'temporarily unavailable' in exc_mess.lower():
 409         return True
 410
 411     return False
 412
 413 class BaseServerTestCase(unittest.TestCase):
 414     requestHandler = None
 415     request_count = 1
 416     threadFunc = staticmethod(http_server)
 417     def setUp(self):
 418         # enable traceback reporting
 419         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True
 420
 421         self.evt = threading.Event()
 422         # start server thread to handle requests
 423         serv_args = (self.evt, self.request_count, self.requestHandler)
 424         threading.Thread(target=self.threadFunc, args=serv_args).start()
 425
 426         # wait for the server to be ready
 427         self.evt.wait(10)
 428         self.evt.clear()
 429
 430     def tearDown(self):
 431         # wait on the server thread to terminate
 432         self.evt.wait(10)
 433
 434         # disable traceback reporting
 435         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = False
 436
 437 # NOTE: The tests in SimpleServerTestCase will ignore failures caused by
 438 # "temporarily unavailable" exceptions raised in SimpleXMLRPCServer.  This
 439 # condition occurs infrequently on some platforms, frequently on others, and
 440 # is apparently caused by using SimpleXMLRPCServer with a non-blocking socket
 441 # If the server class is updated at some point in the future to handle this
 442 # situation more gracefully, these tests should be modified appropriately.
 443
 444 class SimpleServerTestCase(BaseServerTestCase):
 445     def test_simple1(self):
 446         try:
 447             p = xmlrpclib.ServerProxy(URL)
 448             self.assertEqual(p.pow(6,8), 6**8)
 449         except (xmlrpclib.ProtocolError, socket.error), e:
 450             # ignore failures due to non-blocking socket 'unavailable' errors
 451             if not is_unavailable_exception(e):
 452                 # protocol error; provide additional information in test output
 453                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 454
 455     def test_nonascii(self):
 456         start_string = 'P\N{LATIN SMALL LETTER Y WITH CIRCUMFLEX}t'
 457         end_string = 'h\N{LATIN SMALL LETTER O WITH HORN}n'
 458
 459         try:
 460             p = xmlrpclib.ServerProxy(URL)
 461             self.assertEqual(p.add(start_string, end_string),
 462                              start_string + end_string)
 463         except (xmlrpclib.ProtocolError, socket.error) as e:
 464             # ignore failures due to non-blocking socket unavailable errors.
 465             if not is_unavailable_exception(e):
 466                 # protocol error; provide additional information in test output
 467                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 468
 469
 470     # [ch] The test 404 is causing lots of false alarms.
 471     def XXXtest_404(self):
 472         # send POST with httplib, it should return 404 header and
 473         # 'Not Found' message.
 474         conn = httplib.HTTPConnection(ADDR, PORT)
 475         conn.request('POST', '/this-is-not-valid')
 476         response = conn.getresponse()
 477         conn.close()
 478
 479         self.assertEqual(response.status, 404)
 480         self.assertEqual(response.reason, 'Not Found')
 481
 482     def test_introspection1(self):
 483         try:
 484             p = xmlrpclib.ServerProxy(URL)
 485             meth = p.system.listMethods()
 486             expected_methods = set(['pow', 'div', 'my_function', 'add',
 487                                     'system.listMethods', 'system.methodHelp',
 488                                     'system.methodSignature', 'system.multicall'])
 489             self.assertEqual(set(meth), expected_methods)
 490         except (xmlrpclib.ProtocolError, socket.error), e:
 491             # ignore failures due to non-blocking socket 'unavailable' errors
 492             if not is_unavailable_exception(e):
 493                 # protocol error; provide additional information in test output
 494                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 495
 496     def test_introspection2(self):
 497         try:
 498             # test _methodHelp()
 499             p = xmlrpclib.ServerProxy(URL)
 500             divhelp = p.system.methodHelp('div')
 501             self.assertEqual(divhelp, 'This is the div function')
 502         except (xmlrpclib.ProtocolError, socket.error), e:
 503             # ignore failures due to non-blocking socket 'unavailable' errors
 504             if not is_unavailable_exception(e):
 505                 # protocol error; provide additional information in test output
 506                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 507
 508     def test_introspection3(self):
 509         try:
 510             # test native doc
 511             p = xmlrpclib.ServerProxy(URL)
 512             myfunction = p.system.methodHelp('my_function')
 513             self.assertEqual(myfunction, 'This is my function')
 514         except (xmlrpclib.ProtocolError, socket.error), e:
 515             # ignore failures due to non-blocking socket 'unavailable' errors
 516             if not is_unavailable_exception(e):
 517                 # protocol error; provide additional information in test output
 518                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 519
 520     def test_introspection4(self):
 521         # the SimpleXMLRPCServer doesn't support signatures, but
 522         # at least check that we can try making the call
 523         try:
 524             p = xmlrpclib.ServerProxy(URL)
 525             divsig = p.system.methodSignature('div')
 526             self.assertEqual(divsig, 'signatures not supported')
 527         except (xmlrpclib.ProtocolError, socket.error), e:
 528             # ignore failures due to non-blocking socket 'unavailable' errors
 529             if not is_unavailable_exception(e):
 530                 # protocol error; provide additional information in test output
 531                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 532
 533     def test_multicall(self):
 534         try:
 535             p = xmlrpclib.ServerProxy(URL)
 536             multicall = xmlrpclib.MultiCall(p)
 537             multicall.add(2,3)
 538             multicall.pow(6,8)
 539             multicall.div(127,42)
 540             add_result, pow_result, div_result = multicall()
 541             self.assertEqual(add_result, 2+3)
 542             self.assertEqual(pow_result, 6**8)
 543             self.assertEqual(div_result, 127//42)
 544         except (xmlrpclib.ProtocolError, socket.error), e:
 545             # ignore failures due to non-blocking socket 'unavailable' errors
 546             if not is_unavailable_exception(e):
 547                 # protocol error; provide additional information in test output
 548                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 549
 550     def test_non_existing_multicall(self):
 551         try:
 552             p = xmlrpclib.ServerProxy(URL)
 553             multicall = xmlrpclib.MultiCall(p)
 554             multicall.this_is_not_exists()
 555             result = multicall()
 556
 557             # result.results contains;
 558             # [{'faultCode': 1, 'faultString': '<type \'exceptions.Exception\'>:'
 559             #   'method "this_is_not_exists" is not supported'>}]
 560
 561             self.assertEqual(result.results[0]['faultCode'], 1)
 562             self.assertEqual(result.results[0]['faultString'],
 563                 '<type \'exceptions.Exception\'>:method "this_is_not_exists" '
 564                 'is not supported')
 565         except (xmlrpclib.ProtocolError, socket.error), e:
 566             # ignore failures due to non-blocking socket 'unavailable' errors
 567             if not is_unavailable_exception(e):
 568                 # protocol error; provide additional information in test output
 569                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 570
 571     def test_dotted_attribute(self):
 572         # Raises an AttributeError because private methods are not allowed.
 573         self.assertRaises(AttributeError,
 574                           SimpleXMLRPCServer.resolve_dotted_attribute, str, '__add')
 575
 576         self.assertTrue(SimpleXMLRPCServer.resolve_dotted_attribute(str, 'title'))
 577         # Get the test to run faster by sending a request with test_simple1.
 578         # This avoids waiting for the socket timeout.
 579         self.test_simple1()
 580
 581 class MultiPathServerTestCase(BaseServerTestCase):
 582     threadFunc = staticmethod(http_multi_server)
 583     request_count = 2
 584     def test_path1(self):
 585         p = xmlrpclib.ServerProxy(URL+"/foo")
 586         self.assertEqual(p.pow(6,8), 6**8)
 587         self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
 588     def test_path2(self):
 589         p = xmlrpclib.ServerProxy(URL+"/foo/bar")
 590         self.assertEqual(p.add(6,8), 6+8)
 591         self.assertRaises(xmlrpclib.Fault, p.pow, 6, 8)
 592
 593 #A test case that verifies that a server using the HTTP/1.1 keep-alive mechanism
 594 #does indeed serve subsequent requests on the same connection
 595 class BaseKeepaliveServerTestCase(BaseServerTestCase):
 596     #a request handler that supports keep-alive and logs requests into a
 597     #class variable
 598     class RequestHandler(SimpleXMLRPCServer.SimpleXMLRPCRequestHandler):
 599         parentClass = SimpleXMLRPCServer.SimpleXMLRPCRequestHandler
 600         protocol_version = 'HTTP/1.1'
 601         myRequests = []
 602         def handle(self):
 603             self.myRequests.append([])
 604             self.reqidx = len(self.myRequests)-1
 605             return self.parentClass.handle(self)
 606         def handle_one_request(self):
 607             result = self.parentClass.handle_one_request(self)
 608             self.myRequests[self.reqidx].append(self.raw_requestline)
 609             return result
 610
 611     requestHandler = RequestHandler
 612     def setUp(self):
 613         #clear request log
 614         self.RequestHandler.myRequests = []
 615         return BaseServerTestCase.setUp(self)
 616
 617 #A test case that verifies that a server using the HTTP/1.1 keep-alive mechanism
 618 #does indeed serve subsequent requests on the same connection
 619 class KeepaliveServerTestCase1(BaseKeepaliveServerTestCase):
 620     def test_two(self):
 621         p = xmlrpclib.ServerProxy(URL)
 622         #do three requests.
 623         self.assertEqual(p.pow(6,8), 6**8)
 624         self.assertEqual(p.pow(6,8), 6**8)
 625         self.assertEqual(p.pow(6,8), 6**8)
 626
 627         #they should have all been handled by a single request handler
 628         self.assertEqual(len(self.RequestHandler.myRequests), 1)
 629
 630         #check that we did at least two (the third may be pending append
 631         #due to thread scheduling)
 632         self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2)
 633
 634 #test special attribute access on the serverproxy, through the __call__
 635 #function.
 636 class KeepaliveServerTestCase2(BaseKeepaliveServerTestCase):
 637     #ask for two keepalive requests to be handled.
 638     request_count=2
 639
 640     def test_close(self):
 641         p = xmlrpclib.ServerProxy(URL)
 642         #do some requests with close.
 643         self.assertEqual(p.pow(6,8), 6**8)
 644         self.assertEqual(p.pow(6,8), 6**8)
 645         self.assertEqual(p.pow(6,8), 6**8)
 646         p("close")() #this should trigger a new keep-alive request
 647         self.assertEqual(p.pow(6,8), 6**8)
 648         self.assertEqual(p.pow(6,8), 6**8)
 649         self.assertEqual(p.pow(6,8), 6**8)
 650
 651         #they should have all been two request handlers, each having logged at least
 652         #two complete requests
 653         self.assertEqual(len(self.RequestHandler.myRequests), 2)
 654         self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2)
 655         self.assertGreaterEqual(len(self.RequestHandler.myRequests[-2]), 2)
 656
 657     def test_transport(self):
 658         p = xmlrpclib.ServerProxy(URL)
 659         #do some requests with close.
 660         self.assertEqual(p.pow(6,8), 6**8)
 661         p("transport").close() #same as above, really.
 662         self.assertEqual(p.pow(6,8), 6**8)
 663         self.assertEqual(len(self.RequestHandler.myRequests), 2)
 664
 665 #A test case that verifies that gzip encoding works in both directions
 666 #(for a request and the response)
 667 class GzipServerTestCase(BaseServerTestCase):
 668     #a request handler that supports keep-alive and logs requests into a
 669     #class variable
 670     class RequestHandler(SimpleXMLRPCServer.SimpleXMLRPCRequestHandler):
 671         parentClass = SimpleXMLRPCServer.SimpleXMLRPCRequestHandler
 672         protocol_version = 'HTTP/1.1'
 673
 674         def do_POST(self):
 675             #store content of last request in class
 676             self.__class__.content_length = int(self.headers["content-length"])
 677             return self.parentClass.do_POST(self)
 678     requestHandler = RequestHandler
 679
 680     class Transport(xmlrpclib.Transport):
 681         #custom transport, stores the response length for our perusal
 682         fake_gzip = False
 683         def parse_response(self, response):
 684             self.response_length=int(response.getheader("content-length", 0))
 685             return xmlrpclib.Transport.parse_response(self, response)
 686
 687         def send_content(self, connection, body):
 688             if self.fake_gzip:
 689                 #add a lone gzip header to induce decode error remotely
 690                 connection.putheader("Content-Encoding", "gzip")
 691             return xmlrpclib.Transport.send_content(self, connection, body)
 692
 693     def test_gzip_request(self):
 694         t = self.Transport()
 695         t.encode_threshold = None
 696         p = xmlrpclib.ServerProxy(URL, transport=t)
 697         self.assertEqual(p.pow(6,8), 6**8)
 698         a = self.RequestHandler.content_length
 699         t.encode_threshold = 0 #turn on request encoding
 700         self.assertEqual(p.pow(6,8), 6**8)
 701         b = self.RequestHandler.content_length
 702         self.assertTrue(a>b)
 703
 704     def test_bad_gzip_request(self):
 705         t = self.Transport()
 706         t.encode_threshold = None
 707         t.fake_gzip = True
 708         p = xmlrpclib.ServerProxy(URL, transport=t)
 709         cm = self.assertRaisesRegexp(xmlrpclib.ProtocolError,
 710                                      re.compile(r"\b400\b"))
 711         with cm:
 712             p.pow(6, 8)
 713
 714     def test_gsip_response(self):
 715         t = self.Transport()
 716         p = xmlrpclib.ServerProxy(URL, transport=t)
 717         old = self.requestHandler.encode_threshold
 718         self.requestHandler.encode_threshold = None #no encoding
 719         self.assertEqual(p.pow(6,8), 6**8)
 720         a = t.response_length
 721         self.requestHandler.encode_threshold = 0 #always encode
 722         self.assertEqual(p.pow(6,8), 6**8)
 723         b = t.response_length
 724         self.requestHandler.encode_threshold = old
 725         self.assertTrue(a>b)
 726
 727 #Test special attributes of the ServerProxy object
 728 class ServerProxyTestCase(unittest.TestCase):
 729     def test_close(self):
 730         p = xmlrpclib.ServerProxy(URL)
 731         self.assertEqual(p('close')(), None)
 732
 733     def test_transport(self):
 734         t = xmlrpclib.Transport()
 735         p = xmlrpclib.ServerProxy(URL, transport=t)
 736         self.assertEqual(p('transport'), t)
 737
 738 # This is a contrived way to make a failure occur on the server side
 739 # in order to test the _send_traceback_header flag on the server
 740 class FailingMessageClass(mimetools.Message):
 741     def __getitem__(self, key):
 742         key = key.lower()
 743         if key == 'content-length':
 744             return 'I am broken'
 745         return mimetools.Message.__getitem__(self, key)
 746
 747
 748 class FailingServerTestCase(unittest.TestCase):
 749     def setUp(self):
 750         self.evt = threading.Event()
 751         # start server thread to handle requests
 752         serv_args = (self.evt, 1)
 753         threading.Thread(target=http_server, args=serv_args).start()
 754
 755         # wait for the server to be ready
 756         self.evt.wait()
 757         self.evt.clear()
 758
 759     def tearDown(self):
 760         # wait on the server thread to terminate
 761         self.evt.wait()
 762         # reset flag
 763         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = False
 764         # reset message class
 765         SimpleXMLRPCServer.SimpleXMLRPCRequestHandler.MessageClass = mimetools.Message
 766
 767     def test_basic(self):
 768         # check that flag is false by default
 769         flagval = SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header
 770         self.assertEqual(flagval, False)
 771
 772         # enable traceback reporting
 773         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True
 774
 775         # test a call that shouldn't fail just as a smoke test
 776         try:
 777             p = xmlrpclib.ServerProxy(URL)
 778             self.assertEqual(p.pow(6,8), 6**8)
 779         except (xmlrpclib.ProtocolError, socket.error), e:
 780             # ignore failures due to non-blocking socket 'unavailable' errors
 781             if not is_unavailable_exception(e):
 782                 # protocol error; provide additional information in test output
 783                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
 784
 785     def test_fail_no_info(self):
 786         # use the broken message class
 787         SimpleXMLRPCServer.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass
 788
 789         try:
 790             p = xmlrpclib.ServerProxy(URL)
 791             p.pow(6,8)
 792         except (xmlrpclib.ProtocolError, socket.error), e:
 793             # ignore failures due to non-blocking socket 'unavailable' errors
 794             if not is_unavailable_exception(e) and hasattr(e, "headers"):
 795                 # The two server-side error headers shouldn't be sent back in this case
 796                 self.assertTrue(e.headers.get("X-exception") is None)
 797                 self.assertTrue(e.headers.get("X-traceback") is None)
 798         else:
 799             self.fail('ProtocolError not raised')
 800
 801     def test_fail_with_info(self):
 802         # use the broken message class
 803         SimpleXMLRPCServer.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass
 804
 805         # Check that errors in the server send back exception/traceback
 806         # info when flag is set
 807         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True
 808
 809         try:
 810             p = xmlrpclib.ServerProxy(URL)
 811             p.pow(6,8)
 812         except (xmlrpclib.ProtocolError, socket.error), e:
 813             # ignore failures due to non-blocking socket 'unavailable' errors
 814             if not is_unavailable_exception(e) and hasattr(e, "headers"):
 815                 # We should get error info in the response
 816                 expected_err = "invalid literal for int() with base 10: 'I am broken'"
 817                 self.assertEqual(e.headers.get("x-exception"), expected_err)
 818                 self.assertTrue(e.headers.get("x-traceback") is not None)
 819         else:
 820             self.fail('ProtocolError not raised')
 821
 822 class CGIHandlerTestCase(unittest.TestCase):
 823     def setUp(self):
 824         self.cgi = SimpleXMLRPCServer.CGIXMLRPCRequestHandler()
 825
 826     def tearDown(self):
 827         self.cgi = None
 828
 829     def test_cgi_get(self):
 830         with test_support.EnvironmentVarGuard() as env:
 831             env['REQUEST_METHOD'] = 'GET'
 832             # if the method is GET and no request_text is given, it runs handle_get
 833             # get sysout output
 834             with test_support.captured_stdout() as data_out:
 835                 self.cgi.handle_request()
 836
 837             # parse Status header
 838             data_out.seek(0)
 839             handle = data_out.read()
 840             status = handle.split()[1]
 841             message = ' '.join(handle.split()[2:4])
 842
 843             self.assertEqual(status, '400')
 844             self.assertEqual(message, 'Bad Request')
 845
 846
 847     def test_cgi_xmlrpc_response(self):
 848         data = """<?xml version='1.0'?>
 849         <methodCall>
 850             <methodName>test_method</methodName>
 851             <params>
 852                 <param>
 853                     <value><string>foo</string></value>
 854                 </param>
 855                 <param>
 856                     <value><string>bar</string></value>
 857                 </param>
 858             </params>
 859         </methodCall>
 860         """
 861
 862         with test_support.EnvironmentVarGuard() as env, \
 863              test_support.captured_stdout() as data_out, \
 864              test_support.captured_stdin() as data_in:
 865             data_in.write(data)
 866             data_in.seek(0)
 867             env['CONTENT_LENGTH'] = str(len(data))
 868             self.cgi.handle_request()
 869         data_out.seek(0)
 870
 871         # will respond exception, if so, our goal is achieved ;)
 872         handle = data_out.read()
 873
 874         # start with 44th char so as not to get http header, we just need only xml
 875         self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, handle[44:])
 876
 877         # Also test the content-length returned  by handle_request
 878         # Using the same test method inorder to avoid all the datapassing
 879         # boilerplate code.
 880         # Test for bug: http://bugs.python.org/issue5040
 881
 882         content = handle[handle.find("<?xml"):]
 883
 884         self.assertEquals(
 885             int(re.search('Content-Length: (\d+)', handle).group(1)),
 886             len(content))
 887
 888
 889 class FakeSocket:
 890
 891     def __init__(self):
 892         self.data = StringIO.StringIO()
 893
 894     def send(self, buf):
 895         self.data.write(buf)
 896         return len(buf)
 897
 898     def sendall(self, buf):
 899         self.data.write(buf)
 900
 901     def getvalue(self):
 902         return self.data.getvalue()
 903
 904     def makefile(self, x='r', y=-1):
 905         raise RuntimeError
 906
 907     def close(self):
 908         pass
 909
 910 class FakeTransport(xmlrpclib.Transport):
 911     """A Transport instance that records instead of sending a request.
 912
 913     This class replaces the actual socket used by httplib with a
 914     FakeSocket object that records the request.  It doesn't provide a
 915     response.
 916     """
 917
 918     def make_connection(self, host):
 919         conn = xmlrpclib.Transport.make_connection(self, host)
 920         conn.sock = self.fake_socket = FakeSocket()
 921         return conn
 922
 923 class TransportSubclassTestCase(unittest.TestCase):
 924
 925     def issue_request(self, transport_class):
 926         """Return an HTTP request made via transport_class."""
 927         transport = transport_class()
 928         proxy = xmlrpclib.ServerProxy("http://example.com/",
 929                                       transport=transport)
 930         try:
 931             proxy.pow(6, 8)
 932         except RuntimeError:
 933             return transport.fake_socket.getvalue()
 934         return None
 935
 936     def test_custom_user_agent(self):
 937         class TestTransport(FakeTransport):
 938
 939             def send_user_agent(self, conn):
 940                 xmlrpclib.Transport.send_user_agent(self, conn)
 941                 conn.putheader("X-Test", "test_custom_user_agent")
 942
 943         req = self.issue_request(TestTransport)
 944         self.assertTrue("X-Test: test_custom_user_agent\r\n" in req)
 945
 946     def test_send_host(self):
 947         class TestTransport(FakeTransport):
 948
 949             def send_host(self, conn, host):
 950                 xmlrpclib.Transport.send_host(self, conn, host)
 951                 conn.putheader("X-Test", "test_send_host")
 952
 953         req = self.issue_request(TestTransport)
 954         self.assertTrue("X-Test: test_send_host\r\n" in req)
 955
 956     def test_send_request(self):
 957         class TestTransport(FakeTransport):
 958
 959             def send_request(self, conn, url, body):
 960                 xmlrpclib.Transport.send_request(self, conn, url, body)
 961                 conn.putheader("X-Test", "test_send_request")
 962
 963         req = self.issue_request(TestTransport)
 964         self.assertTrue("X-Test: test_send_request\r\n" in req)
 965
 966     def test_send_content(self):
 967         class TestTransport(FakeTransport):
 968
 969             def send_content(self, conn, body):
 970                 conn.putheader("X-Test", "test_send_content")
 971                 xmlrpclib.Transport.send_content(self, conn, body)
 972
 973         req = self.issue_request(TestTransport)
 974         self.assertTrue("X-Test: test_send_content\r\n" in req)
 975
 976 @test_support.reap_threads
 977 def test_main():
 978     xmlrpc_tests = [XMLRPCTestCase, HelperTestCase, DateTimeTestCase,
 979          BinaryTestCase, FaultTestCase, TransportSubclassTestCase]
 980     xmlrpc_tests.append(SimpleServerTestCase)
 981     xmlrpc_tests.append(KeepaliveServerTestCase1)
 982     xmlrpc_tests.append(KeepaliveServerTestCase2)
 983     try:
 984         import gzip
 985         xmlrpc_tests.append(GzipServerTestCase)
 986     except ImportError:
 987         pass #gzip not supported in this build
 988     xmlrpc_tests.append(MultiPathServerTestCase)
 989     xmlrpc_tests.append(ServerProxyTestCase)
 990     xmlrpc_tests.append(FailingServerTestCase)
 991     xmlrpc_tests.append(CGIHandlerTestCase)
 992
 993     test_support.run_unittest(*xmlrpc_tests)
 994
 995 if __name__ == "__main__":
 996     test_main()