~ubuntu-branches/ubuntu/precise/python3.2/precise-proposed

« back to all changes in this revision

Viewing changes to Lib/test/test_xmlrpc.py

  • Committer: Package Import Robot
  • Author(s): Matthias Klose
  • Date: 2012-03-09 18:40:39 UTC
  • mfrom: (30.1.2 sid)
  • Revision ID: package-import@ubuntu.com-20120309184039-j3yk2emxr1plyo21
Tags: 3.2.3~rc1-1
* Python 3.2.3 release candidate 1.
* Update to 20120309 from the 3.2 branch.
* Fix libpython.a symlink. Closes: #660146.
* Build-depend on xauth.
* Run the gdb tests for the debug build only.

Show diffs side-by-side

added added

removed removed

Lines of Context:
66
66
        (newdt,), m = xmlrpclib.loads(s, use_datetime=0)
67
67
        self.assertEqual(newdt, xmlrpclib.DateTime('00010210T11:41:23'))
68
68
 
69
 
    def test_cmp_datetime_DateTime(self):
70
 
        now = datetime.datetime.now()
71
 
        dt = xmlrpclib.DateTime(now.timetuple())
72
 
        self.assertTrue(dt == now)
73
 
        self.assertTrue(now == dt)
74
 
        then = now + datetime.timedelta(seconds=4)
75
 
        self.assertTrue(then >= dt)
76
 
        self.assertTrue(dt < then)
77
 
 
78
69
    def test_bug_1164912 (self):
79
70
        d = xmlrpclib.DateTime()
80
71
        ((new_d,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((d,),
149
140
                          ('host.tld',
150
141
                           [('Authorization', 'Basic dXNlcg==')], {}))
151
142
 
 
143
    def test_dump_bytes(self):
 
144
        self.assertRaises(TypeError, xmlrpclib.dumps, (b"my dog has fleas",))
 
145
 
152
146
    def test_ssl_presence(self):
153
147
        try:
154
148
            import ssl
186
180
        self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, s)
187
181
 
188
182
    def test_dotted_attribute(self):
189
 
        # this will raise AttirebuteError because code don't want us to use
 
183
        # this will raise AttributeError because code don't want us to use
190
184
        # private methods
191
185
        self.assertRaises(AttributeError,
192
186
                          xmlrpc.server.resolve_dotted_attribute, str, '__add')
233
227
        t2 = xmlrpclib._datetime(d)
234
228
        self.assertEqual(t1, tref)
235
229
 
 
230
    def test_comparison(self):
 
231
        now = datetime.datetime.now()
 
232
        dtime = xmlrpclib.DateTime(now.timetuple())
 
233
 
 
234
        # datetime vs. DateTime
 
235
        self.assertTrue(dtime == now)
 
236
        self.assertTrue(now == dtime)
 
237
        then = now + datetime.timedelta(seconds=4)
 
238
        self.assertTrue(then >= dtime)
 
239
        self.assertTrue(dtime < then)
 
240
 
 
241
        # str vs. DateTime
 
242
        dstr = now.strftime("%Y%m%dT%H:%M:%S")
 
243
        self.assertTrue(dtime == dstr)
 
244
        self.assertTrue(dstr == dtime)
 
245
        dtime_then = xmlrpclib.DateTime(then.timetuple())
 
246
        self.assertTrue(dtime_then >= dstr)
 
247
        self.assertTrue(dstr < dtime_then)
 
248
 
 
249
        # some other types
 
250
        dbytes = dstr.encode('ascii')
 
251
        dtuple = now.timetuple()
 
252
        with self.assertRaises(TypeError):
 
253
            dtime == 1970
 
254
        with self.assertRaises(TypeError):
 
255
            dtime != dbytes
 
256
        with self.assertRaises(TypeError):
 
257
            dtime == bytearray(dbytes)
 
258
        with self.assertRaises(TypeError):
 
259
            dtime != dtuple
 
260
        with self.assertRaises(TypeError):
 
261
            dtime < float(1970)
 
262
        with self.assertRaises(TypeError):
 
263
            dtime > dbytes
 
264
        with self.assertRaises(TypeError):
 
265
            dtime <= bytearray(dbytes)
 
266
        with self.assertRaises(TypeError):
 
267
            dtime >= dtuple
 
268
 
236
269
class BinaryTestCase(unittest.TestCase):
237
270
 
238
271
    # XXX What should str(Binary(b"\xff")) return?  I'm chosing "\xff"
295
328
        global ADDR, PORT, URL
296
329
        ADDR, PORT = serv.socket.getsockname()
297
330
        #connect to IP address directly.  This avoids socket.create_connection()
298
 
        #trying to connect to to "localhost" using all address families, which
 
331
        #trying to connect to "localhost" using all address families, which
299
332
        #causes slowdown e.g. on vista which supports AF_INET6.  The server listens
300
333
        #on AF_INET only.
301
334
        URL = "http://%s:%d"%(ADDR, PORT)
346
379
    class MyRequestHandler(requestHandler):
347
380
        rpc_paths = []
348
381
 
 
382
    class BrokenDispatcher:
 
383
        def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
 
384
            raise RuntimeError("broken dispatcher")
 
385
 
349
386
    serv = MyXMLRPCServer(("localhost", 0), MyRequestHandler,
350
387
                          logRequests=False, bind_and_activate=False)
351
388
    serv.socket.settimeout(3)
354
391
        global ADDR, PORT, URL
355
392
        ADDR, PORT = serv.socket.getsockname()
356
393
        #connect to IP address directly.  This avoids socket.create_connection()
357
 
        #trying to connect to to "localhost" using all address families, which
 
394
        #trying to connect to "localhost" using all address families, which
358
395
        #causes slowdown e.g. on vista which supports AF_INET6.  The server listens
359
396
        #on AF_INET only.
360
397
        URL = "http://%s:%d"%(ADDR, PORT)
366
403
            d.register_multicall_functions()
367
404
        serv.get_dispatcher(paths[0]).register_function(pow)
368
405
        serv.get_dispatcher(paths[1]).register_function(lambda x,y: x+y, 'add')
 
406
        serv.add_dispatcher("/is/broken", BrokenDispatcher())
369
407
        evt.set()
370
408
 
371
409
        # handle up to 'numrequests' requests
436
474
 
437
475
    def tearDown(self):
438
476
        # wait on the server thread to terminate
439
 
        self.evt.wait(4.0)
440
 
        # XXX this code does not work, and in fact stop_serving doesn't exist.
441
 
        if not self.evt.is_set():
442
 
            self.evt.set()
443
 
            stop_serving()
444
 
            raise RuntimeError("timeout reached, test has failed")
 
477
        self.evt.wait()
445
478
 
446
479
        # disable traceback reporting
447
480
        xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = False
584
617
        # This avoids waiting for the socket timeout.
585
618
        self.test_simple1()
586
619
 
 
620
    def test_unicode_host(self):
 
621
        server = xmlrpclib.ServerProxy("http://%s:%d/RPC2" % (ADDR, PORT))
 
622
        self.assertEqual(server.add("a", "\xe9"), "a\xe9")
 
623
 
 
624
    def test_partial_post(self):
 
625
        # Check that a partial POST doesn't make the server loop: issue #14001.
 
626
        conn = http.client.HTTPConnection(ADDR, PORT)
 
627
        conn.request('POST', '/RPC2 HTTP/1.0\r\nContent-Length: 100\r\n\r\nbye')
 
628
        conn.close()
 
629
 
 
630
 
587
631
class MultiPathServerTestCase(BaseServerTestCase):
588
632
    threadFunc = staticmethod(http_multi_server)
589
633
    request_count = 2
591
635
        p = xmlrpclib.ServerProxy(URL+"/foo")
592
636
        self.assertEqual(p.pow(6,8), 6**8)
593
637
        self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
 
638
 
594
639
    def test_path2(self):
595
640
        p = xmlrpclib.ServerProxy(URL+"/foo/bar")
596
641
        self.assertEqual(p.add(6,8), 6+8)
597
642
        self.assertRaises(xmlrpclib.Fault, p.pow, 6, 8)
598
643
 
 
644
    def test_path3(self):
 
645
        p = xmlrpclib.ServerProxy(URL+"/is/broken")
 
646
        self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
 
647
 
599
648
#A test case that verifies that a server using the HTTP/1.1 keep-alive mechanism
600
649
#does indeed serve subsequent requests on the same connection
601
650
class BaseKeepaliveServerTestCase(BaseServerTestCase):