(newdt,), m = xmlrpclib.loads(s, use_datetime=0)
self.assertEqual(newdt, xmlrpclib.DateTime('00010210T11:41:23'))
- def test_cmp_datetime_DateTime(self):
- now = datetime.datetime.now()
- dt = xmlrpclib.DateTime(now.timetuple())
- self.assertTrue(dt == now)
- self.assertTrue(now == dt)
- then = now + datetime.timedelta(seconds=4)
- self.assertTrue(then >= dt)
- self.assertTrue(dt < then)
-
def test_bug_1164912 (self):
d = xmlrpclib.DateTime()
((new_d,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((d,),
t2 = xmlrpclib._datetime(d)
self.assertEqual(t1, tref)
+ def test_comparison(self):
+ now = datetime.datetime.now()
+ dtime = xmlrpclib.DateTime(now.timetuple())
+
+ # datetime vs. DateTime
+ self.assertTrue(dtime == now)
+ self.assertTrue(now == dtime)
+ then = now + datetime.timedelta(seconds=4)
+ self.assertTrue(then >= dtime)
+ self.assertTrue(dtime < then)
+
+ # str vs. DateTime
+ dstr = now.strftime("%Y%m%dT%H:%M:%S")
+ self.assertTrue(dtime == dstr)
+ self.assertTrue(dstr == dtime)
+ dtime_then = xmlrpclib.DateTime(then.timetuple())
+ self.assertTrue(dtime_then >= dstr)
+ self.assertTrue(dstr < dtime_then)
+
+ # some other types
+ dbytes = dstr.encode('ascii')
+ dtuple = now.timetuple()
+ with self.assertRaises(TypeError):
+ dtime == 1970
+ with self.assertRaises(TypeError):
+ dtime != dbytes
+ with self.assertRaises(TypeError):
+ dtime == bytearray(dbytes)
+ with self.assertRaises(TypeError):
+ dtime != dtuple
+ with self.assertRaises(TypeError):
+ dtime < float(1970)
+ with self.assertRaises(TypeError):
+ dtime > dbytes
+ with self.assertRaises(TypeError):
+ dtime <= bytearray(dbytes)
+ with self.assertRaises(TypeError):
+ dtime >= dtuple
+
class BinaryTestCase(unittest.TestCase):
# XXX What should str(Binary(b"\xff")) return? I'm chosing "\xff"
class MyRequestHandler(requestHandler):
rpc_paths = []
+ class BrokenDispatcher:
+ def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
+ raise RuntimeError("broken dispatcher")
+
serv = MyXMLRPCServer(("localhost", 0), MyRequestHandler,
logRequests=False, bind_and_activate=False)
serv.socket.settimeout(3)
d.register_multicall_functions()
serv.get_dispatcher(paths[0]).register_function(pow)
serv.get_dispatcher(paths[1]).register_function(lambda x,y: x+y, 'add')
+ serv.add_dispatcher("/is/broken", BrokenDispatcher())
evt.set()
# handle up to 'numrequests' requests
p = xmlrpclib.ServerProxy(URL+"/foo")
self.assertEqual(p.pow(6,8), 6**8)
self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
+
def test_path2(self):
p = xmlrpclib.ServerProxy(URL+"/foo/bar")
self.assertEqual(p.add(6,8), 6+8)
self.assertRaises(xmlrpclib.Fault, p.pow, 6, 8)
+ def test_path3(self):
+ p = xmlrpclib.ServerProxy(URL+"/is/broken")
+ self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
+
#A test case that verifies that a server using the HTTP/1.1 keep-alive mechanism
#does indeed serve subsequent requests on the same connection
class BaseKeepaliveServerTestCase(BaseServerTestCase):
encoding, bind_and_activate)
self.dispatchers = {}
self.allow_none = allow_none
- self.encoding = encoding
+ self.encoding = encoding or 'utf-8'
def add_dispatcher(self, path, dispatcher):
self.dispatchers[path] = dispatcher
# (each dispatcher should have handled their own
# exceptions)
exc_type, exc_value = sys.exc_info()[:2]
- response = xmlrpclib.dumps(
- xmlrpclib.Fault(1, "%s:%s" % (exc_type, exc_value)),
+ response = dumps(
+ Fault(1, "%s:%s" % (exc_type, exc_value)),
encoding=self.encoding, allow_none=self.allow_none)
+ response = response.encode(self.encoding)
return response
class CGIXMLRPCRequestHandler(SimpleXMLRPCDispatcher):