66
66
(newdt,), m = xmlrpclib.loads(s, use_datetime=0)
67
67
self.assertEqual(newdt, xmlrpclib.DateTime('00010210T11:41:23'))
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)
78
69
def test_bug_1164912 (self):
79
70
d = xmlrpclib.DateTime()
80
71
((new_d,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((d,),
150
141
[('Authorization', 'Basic dXNlcg==')], {}))
143
def test_dump_bytes(self):
144
self.assertRaises(TypeError, xmlrpclib.dumps, (b"my dog has fleas",))
152
146
def test_ssl_presence(self):
186
180
self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, s)
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)
230
def test_comparison(self):
231
now = datetime.datetime.now()
232
dtime = xmlrpclib.DateTime(now.timetuple())
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)
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)
250
dbytes = dstr.encode('ascii')
251
dtuple = now.timetuple()
252
with self.assertRaises(TypeError):
254
with self.assertRaises(TypeError):
256
with self.assertRaises(TypeError):
257
dtime == bytearray(dbytes)
258
with self.assertRaises(TypeError):
260
with self.assertRaises(TypeError):
262
with self.assertRaises(TypeError):
264
with self.assertRaises(TypeError):
265
dtime <= bytearray(dbytes)
266
with self.assertRaises(TypeError):
236
269
class BinaryTestCase(unittest.TestCase):
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):
382
class BrokenDispatcher:
383
def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
384
raise RuntimeError("broken dispatcher")
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())
371
409
# handle up to 'numrequests' requests
437
475
def tearDown(self):
438
476
# wait on the server thread to terminate
440
# XXX this code does not work, and in fact stop_serving doesn't exist.
441
if not self.evt.is_set():
444
raise RuntimeError("timeout reached, test has failed")
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()
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")
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')
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)
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)
644
def test_path3(self):
645
p = xmlrpclib.ServerProxy(URL+"/is/broken")
646
self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
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):