mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 18:29:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

merging revision 73986 from trunk:

Browse source 
http://bugs.python.org/issue6267
Add more tests for the xlmrpc.ServerProxy
This commit is contained in:
Kristján Valur Jónsson 2009-07-12 22:45:18 +00:00
parent 6db9470efd
commit f6087ca999
1 changed files with 41 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

45
Lib/test/test_xmlrpc.py
View file

@ -322,13 +322,14 @@ def is_unavailable_exception(e):
class BaseServerTestCase(unittest.TestCase): class BaseServerTestCase(unittest.TestCase):
requestHandler = None requestHandler = None
request_count = 1
def setUp(self): def setUp(self):
# enable traceback reporting # enable traceback reporting
xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = True xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = True
self.evt = threading.Event() self.evt = threading.Event()
# start server thread to handle requests # start server thread to handle requests
serv_args = (self.evt, 1, self.requestHandler) serv_args = (self.evt, self.request_count, self.requestHandler)
threading.Thread(target=http_server, args=serv_args).start() threading.Thread(target=http_server, args=serv_args).start()
# wait for the server to be ready # wait for the server to be ready
@ -484,7 +485,7 @@ class SimpleServerTestCase(BaseServerTestCase):
#A test case that verifies that a server using the HTTP/1.1 keep-alive mechanism #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 #does indeed serve subsequent requests on the same connection
class KeepaliveServerTestCase(BaseServerTestCase): class BaseKeepaliveServerTestCase(BaseServerTestCase):
#a request handler that supports keep-alive and logs requests into a #a request handler that supports keep-alive and logs requests into a
#class variable #class variable
class RequestHandler(xmlrpc.server.SimpleXMLRPCRequestHandler): class RequestHandler(xmlrpc.server.SimpleXMLRPCRequestHandler):
@ -493,10 +494,11 @@ class KeepaliveServerTestCase(BaseServerTestCase):
myRequests = [] myRequests = []
def handle(self): def handle(self):
self.myRequests.append([]) self.myRequests.append([])
self.reqidx = len(self.myRequests)-1
return self.parentClass.handle(self) return self.parentClass.handle(self)
def handle_one_request(self): def handle_one_request(self):
result = self.parentClass.handle_one_request(self) result = self.parentClass.handle_one_request(self)
self.myRequests[-1].append(self.raw_requestline) self.myRequests[self.reqidx].append(self.raw_requestline)
return result return result
requestHandler = RequestHandler requestHandler = RequestHandler
@ -505,6 +507,9 @@ class KeepaliveServerTestCase(BaseServerTestCase):
self.RequestHandler.myRequests = [] self.RequestHandler.myRequests = []
return BaseServerTestCase.setUp(self) return BaseServerTestCase.setUp(self)
#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 KeepaliveServerTestCase1(BaseKeepaliveServerTestCase):
def test_two(self): def test_two(self):
p = xmlrpclib.ServerProxy(URL) p = xmlrpclib.ServerProxy(URL)
#do three requests. #do three requests.
@ -519,6 +524,37 @@ class KeepaliveServerTestCase(BaseServerTestCase):
#due to thread scheduling) #due to thread scheduling)
self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2) self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2)
#test special attribute access on the serverproxy, through the __call__
#function.
class KeepaliveServerTestCase2(BaseKeepaliveServerTestCase):
#ask for two keepalive requests to be handled.
request_count=2
def test_close(self):
p = xmlrpclib.ServerProxy(URL)
#do some requests with close.
self.assertEqual(p.pow(6,8), 6**8)
self.assertEqual(p.pow(6,8), 6**8)
self.assertEqual(p.pow(6,8), 6**8)
p("close")() #this should trigger a new keep-alive request
self.assertEqual(p.pow(6,8), 6**8)
self.assertEqual(p.pow(6,8), 6**8)
self.assertEqual(p.pow(6,8), 6**8)
#they should have all been two request handlers, each having logged at least
#two complete requests
self.assertEqual(len(self.RequestHandler.myRequests), 2)
self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2)
self.assertGreaterEqual(len(self.RequestHandler.myRequests[-2]), 2)
def test_transport(self):
p = xmlrpclib.ServerProxy(URL)
#do some requests with close.
self.assertEqual(p.pow(6,8), 6**8)
p("transport").close() #same as above, really.
self.assertEqual(p.pow(6,8), 6**8)
self.assertEqual(len(self.RequestHandler.myRequests), 2)
#A test case that verifies that gzip encoding works in both directions #A test case that verifies that gzip encoding works in both directions
#(for a request and the response) #(for a request and the response)
class GzipServerTestCase(BaseServerTestCase): class GzipServerTestCase(BaseServerTestCase):
@ -761,7 +797,8 @@ def test_main():
xmlrpc_tests = [XMLRPCTestCase, HelperTestCase, DateTimeTestCase, xmlrpc_tests = [XMLRPCTestCase, HelperTestCase, DateTimeTestCase,
BinaryTestCase, FaultTestCase] BinaryTestCase, FaultTestCase]
xmlrpc_tests.append(SimpleServerTestCase) xmlrpc_tests.append(SimpleServerTestCase)
xmlrpc_tests.append(KeepaliveServerTestCase) xmlrpc_tests.append(KeepaliveServerTestCase1)
xmlrpc_tests.append(KeepaliveServerTestCase2)
xmlrpc_tests.append(GzipServerTestCase) xmlrpc_tests.append(GzipServerTestCase)
xmlrpc_tests.append(ServerProxyTestCase) xmlrpc_tests.append(ServerProxyTestCase)
xmlrpc_tests.append(FailingServerTestCase) xmlrpc_tests.append(FailingServerTestCase)