From: Facundo Batista <facundobatista@gmail.com>
Date: Sun, 25 Mar 2007 01:53:21 +0000 (+0000)
Subject: Redone the tests, using the infrastructure already present
X-Git-Tag: v2.6a1~1931
X-Git-Url: https://granicus.if.org/sourcecode?a=commitdiff_plain;h=b8af7bcad55b00234766662aa13032af22906963;p=python

Redone the tests, using the infrastructure already present
for threading and socket serving.
---

diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index ade524f807..5be3dc3b5e 100644
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -820,74 +820,85 @@ class BasicTCPTest2(NetworkConnectionTest, BasicTCPTest):
     """Tests that NetworkConnection does not break existing TCP functionality.
     """
 
-class NetworkConnectionAttributesTest(unittest.TestCase):
+class NetworkConnectionNoServer(unittest.TestCase):
+    def testWithoutServer(self):
+        self.failUnlessRaises(socket.error, lambda: socket.create_connection((HOST, PORT)))
 
-    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)
+class NetworkConnectionAttributesTest(SocketTCPTest, ThreadableTest):
 
-    def tearDown(self):
-        if self.serv:
-            self.serv.close()
-            self.serv = None
+    def __init__(self, methodName='runTest'):
+        SocketTCPTest.__init__(self, methodName=methodName)
+        ThreadableTest.__init__(self)
 
-    def testWithoutServer(self):
-        self.tearDown()
-        self.failUnlessRaises(socket.error, lambda: socket.create_connection((HOST, PORT)))
+    def clientSetUp(self):
+        pass
 
-    def testTimeoutAttribute(self):
-        # default
-        sock = socket.create_connection((HOST, PORT))
-        self.assertTrue(sock.gettimeout() is None)
-    
-        # a value, named
-        sock = socket.create_connection((HOST, PORT), timeout=30)
-        self.assertEqual(sock.gettimeout(), 30)
+    def clientTearDown(self):
+        self.cli.close()
+        self.cli = None
+        ThreadableTest.clientTearDown(self)
 
-        # a value, just the value
-        sock = socket.create_connection((HOST, PORT), 30)
-        self.assertEqual(sock.gettimeout(), 30)
+    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)
 
-        # None, having other default 
+    testTimeoutDefault = _justAccept 
+    def _testTimeoutDefault(self):
+        self.cli = socket.create_connection((HOST, PORT))
+        self.assertTrue(self.cli.gettimeout() is None)
+    
+    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)
+
+    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)
-
-    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):
+        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 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 @@ def test_main():
         LineBufferedFileObjectClassTestCase,
         SmallBufferedFileObjectClassTestCase,
         Urllib2FileobjectTest,
+        NetworkConnectionNoServer,
         NetworkConnectionAttributesTest,
         NetworkConnectionBehaviourTest,
     ])