mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-12-09 02:35:14 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

test_xmlrpc: close the transport when done

Browse source 
Fix a ResourceWarning(unclosed socket). Patch written by Nadeem Vawda.
This commit is contained in:
Victor Stinner 2011-01-03 14:30:44 +00:00
parent 270fe40831
commit a935e8ffc6
1 changed files with 8 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

8
Lib/test/test_xmlrpc.py
View file

@ -629,6 +629,7 @@ class KeepaliveServerTestCase1(BaseKeepaliveServerTestCase):
self.assertEqual(p.pow(6,8), 6**8) self.assertEqual(p.pow(6,8), 6**8)
self.assertEqual(p.pow(6,8), 6**8) 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")()
#they should have all been handled by a single request handler #they should have all been handled by a single request handler
self.assertEqual(len(self.RequestHandler.myRequests), 1) self.assertEqual(len(self.RequestHandler.myRequests), 1)
@ -637,6 +638,7 @@ class KeepaliveServerTestCase1(BaseKeepaliveServerTestCase):
#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__ #test special attribute access on the serverproxy, through the __call__
#function. #function.
class KeepaliveServerTestCase2(BaseKeepaliveServerTestCase): class KeepaliveServerTestCase2(BaseKeepaliveServerTestCase):
@ -653,6 +655,7 @@ class KeepaliveServerTestCase2(BaseKeepaliveServerTestCase):
self.assertEqual(p.pow(6,8), 6**8) self.assertEqual(p.pow(6,8), 6**8)
self.assertEqual(p.pow(6,8), 6**8) 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")()
#they should have all been two request handlers, each having logged at least #they should have all been two request handlers, each having logged at least
#two complete requests #two complete requests
@ -660,12 +663,14 @@ class KeepaliveServerTestCase2(BaseKeepaliveServerTestCase):
self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2) self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2)
self.assertGreaterEqual(len(self.RequestHandler.myRequests[-2]), 2) self.assertGreaterEqual(len(self.RequestHandler.myRequests[-2]), 2)
def test_transport(self): def test_transport(self):
p = xmlrpclib.ServerProxy(URL) p = xmlrpclib.ServerProxy(URL)
#do some requests with close. #do some requests with close.
self.assertEqual(p.pow(6,8), 6**8) self.assertEqual(p.pow(6,8), 6**8)
p("transport").close() #same as above, really. p("transport").close() #same as above, really.
self.assertEqual(p.pow(6,8), 6**8) self.assertEqual(p.pow(6,8), 6**8)
p("close")()
self.assertEqual(len(self.RequestHandler.myRequests), 2) 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
@ -709,6 +714,7 @@ class GzipServerTestCase(BaseServerTestCase):
self.assertEqual(p.pow(6,8), 6**8) self.assertEqual(p.pow(6,8), 6**8)
b = self.RequestHandler.content_length b = self.RequestHandler.content_length
self.assertTrue(a>b) self.assertTrue(a>b)
p("close")()
def test_bad_gzip_request(self): def test_bad_gzip_request(self):
t = self.Transport() t = self.Transport()
@ -719,6 +725,7 @@ class GzipServerTestCase(BaseServerTestCase):
re.compile(r"\b400\b")) re.compile(r"\b400\b"))
with cm: with cm:
p.pow(6, 8) p.pow(6, 8)
p("close")()
def test_gsip_response(self): def test_gsip_response(self):
t = self.Transport() t = self.Transport()
@ -729,6 +736,7 @@ class GzipServerTestCase(BaseServerTestCase):
a = t.response_length a = t.response_length
self.requestHandler.encode_threshold = 0 #always encode self.requestHandler.encode_threshold = 0 #always encode
self.assertEqual(p.pow(6,8), 6**8) self.assertEqual(p.pow(6,8), 6**8)
p("close")()
b = t.response_length b = t.response_length
self.requestHandler.encode_threshold = old self.requestHandler.encode_threshold = old
self.assertTrue(a>b) self.assertTrue(a>b)