[Python-checkins] r54568 - python/trunk/Lib/test/test_socket.py

Sun Mar 25 03:53:25 CEST 2007

Author: facundo.batista
Date: Sun Mar 25 03:53:21 2007
New Revision: 54568
Modified:
 python/trunk/Lib/test/test_socket.py
Log:
Redone the tests, using the infrastructure already present
for threading and socket serving.
Modified: python/trunk/Lib/test/test_socket.py
==============================================================================

--- python/trunk/Lib/test/test_socket.py	(original)
+++ python/trunk/Lib/test/test_socket.py	Sun Mar 25 03:53:21 2007
@@ -820,74 +820,85 @@
 """Tests that NetworkConnection does not break existing TCP functionality.
 """
 
-class NetworkConnectionAttributesTest(unittest.TestCase):
-
- def setUp(self):
- self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- global PORT
- PORT = test_support.bind_port(self.serv, HOST, PORT)
- self.serv.listen(1)
-
- def tearDown(self):
- if self.serv:
- self.serv.close()
- self.serv = None
-
+class NetworkConnectionNoServer(unittest.TestCase):
 def testWithoutServer(self):
- self.tearDown()
 self.failUnlessRaises(socket.error, lambda: socket.create_connection((HOST, PORT)))
 
- def testTimeoutAttribute(self):
- # default
- sock = socket.create_connection((HOST, PORT))
- self.assertTrue(sock.gettimeout() is None)
+class NetworkConnectionAttributesTest(SocketTCPTest, ThreadableTest):
+
+ def __init__(self, methodName='runTest'):
+ SocketTCPTest.__init__(self, methodName=methodName)
+ ThreadableTest.__init__(self)
+
+ def clientSetUp(self):
+ pass
+
+ def clientTearDown(self):
+ self.cli.close()
+ self.cli = None
+ ThreadableTest.clientTearDown(self)
+
+ def _justAccept(self):
+ conn, addr = self.serv.accept()
+
+ testFamily = _justAccept
+ def _testFamily(self):
+ self.cli = socket.create_connection((HOST, PORT), timeout=30)
+ self.assertEqual(self.cli.family, 2)
+
+ testTimeoutDefault = _justAccept 
+ def _testTimeoutDefault(self):
+ self.cli = socket.create_connection((HOST, PORT))
+ self.assertTrue(self.cli.gettimeout() is None)
 
- # a value, named
- sock = socket.create_connection((HOST, PORT), timeout=30)
- self.assertEqual(sock.gettimeout(), 30)
-
- # a value, just the value
- sock = socket.create_connection((HOST, PORT), 30)
- self.assertEqual(sock.gettimeout(), 30)
+ testTimeoutValueNamed = _justAccept 
+ def _testTimeoutValueNamed(self):
+ self.cli = socket.create_connection((HOST, PORT), timeout=30)
+ self.assertEqual(self.cli.gettimeout(), 30)
+
+ testTimeoutValueNonamed = _justAccept 
+ def _testTimeoutValueNonamed(self):
+ self.cli = socket.create_connection((HOST, PORT), 30)
+ self.assertEqual(self.cli.gettimeout(), 30)
 
- # None, having other default 
+ testTimeoutNone = _justAccept 
+ def _testTimeoutNone(self):
 previous = socket.getdefaulttimeout()
 socket.setdefaulttimeout(30)
 try:
- sock = socket.create_connection((HOST, PORT), timeout=None)
+ self.cli = socket.create_connection((HOST, PORT), timeout=None)
 finally:
 socket.setdefaulttimeout(previous)
- self.assertEqual(sock.gettimeout(), 30)
+ self.assertEqual(self.cli.gettimeout(), 30)
+
+
+class NetworkConnectionBehaviourTest(SocketTCPTest, ThreadableTest):
+
+ def __init__(self, methodName='runTest'):
+ SocketTCPTest.__init__(self, methodName=methodName)
+ ThreadableTest.__init__(self)
+
+ def clientSetUp(self):
+ pass
+
+ def clientTearDown(self):
+ self.cli.close()
+ self.cli = None
+ ThreadableTest.clientTearDown(self)
 
- def testFamily(self):
- sock = socket.create_connection((HOST, PORT), timeout=30)
- self.assertEqual(sock.family, 2)
-
-
-def threadedServer(delay):
- serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- global PORT
- PORT = test_support.bind_port(serv, HOST, PORT)
- serv.listen(1)
- conn, addr = serv.accept()
- time.sleep(delay)
- conn.send("done!")
- conn.close()
- 
-class NetworkConnectionBehaviourTest(unittest.TestCase):
 def testInsideTimeout(self):
- threading.Thread(target=threadedServer, args=(3,)).start()
- time.sleep(.1)
- sock = socket.create_connection((HOST, PORT))
+ conn, addr = self.serv.accept()
+ time.sleep(3)
+ conn.send("done!")
+ testOutsideTimeout = testInsideTimeout
+
+ def _testInsideTimeout(self):
+ self.cli = sock = socket.create_connection((HOST, PORT))
 data = sock.recv(5)
 self.assertEqual(data, "done!")
 
- def testOutsideTimeout(self):
- threading.Thread(target=threadedServer, args=(3,)).start()
- time.sleep(.1)
- sock = socket.create_connection((HOST, PORT), timeout=1)
+ def _testOutsideTimeout(self):
+ self.cli = sock = socket.create_connection((HOST, PORT), timeout=1)
 self.failUnlessRaises(socket.timeout, lambda: sock.recv(5))
 
 
@@ -1069,6 +1080,7 @@
 LineBufferedFileObjectClassTestCase,
 SmallBufferedFileObjectClassTestCase,
 Urllib2FileobjectTest,
+ NetworkConnectionNoServer,
 NetworkConnectionAttributesTest,
 NetworkConnectionBehaviourTest,
 ])
