]> granicus.if.org Git - python/commitdiff
test_timeout: move testRecvfromTimeout() to a UDP-specific test case
authorVictor Stinner <victor.stinner@haypocalc.com>
Mon, 3 Jan 2011 14:30:41 +0000 (14:30 +0000)
committerVictor Stinner <victor.stinner@haypocalc.com>
Mon, 3 Jan 2011 14:30:41 +0000 (14:30 +0000)
Fix a ResourceWarning(unclosed socket).

Lib/test/test_timeout.py

index bbcdc25672ac56cb7da8f28994649a5f7fb3992b..618b0e4a81e71fb7b1ffa48110cee4fd0683a300 100644 (file)
@@ -88,8 +88,6 @@ class CreationTestCase(unittest.TestCase):
 
 
 class TimeoutTestCase(unittest.TestCase):
-    """Test case for socket.socket() timeout functions"""
-
     # There are a number of tests here trying to make sure that an operation
     # doesn't take too much longer than expected.  But competing machine
     # activity makes it inevitable that such tests will fail at times.
@@ -98,14 +96,22 @@ class TimeoutTestCase(unittest.TestCase):
     # solution.
     fuzz = 2.0
 
+    localhost = '127.0.0.1'
+
     def setUp(self):
-        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.addr_remote = ('www.python.org.', 80)
-        self.localhost = '127.0.0.1'
+        raise NotImplementedError()
 
     def tearDown(self):
         self.sock.close()
 
+
+class TCPTimeoutTestCase(TimeoutTestCase):
+    """TCP test case for socket.socket() timeout functions"""
+
+    def setUp(self):
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.addr_remote = ('www.python.org.', 80)
+
     def testConnectTimeout(self):
         # Choose a private address that is unlikely to exist to prevent
         # failures due to the connect succeeding before the timeout.
@@ -161,10 +167,31 @@ class TimeoutTestCase(unittest.TestCase):
                      "timeout (%g) is %g seconds more than expected (%g)"
                      %(_delta, self.fuzz, _timeout))
 
+    def testSend(self):
+        # Test send() timeout
+        # couldn't figure out how to test it
+        pass
+
+    def testSendto(self):
+        # Test sendto() timeout
+        # couldn't figure out how to test it
+        pass
+
+    def testSendall(self):
+        # Test sendall() timeout
+        # couldn't figure out how to test it
+        pass
+
+
+class UDPTimeoutTestCase(TimeoutTestCase):
+    """UDP test case for socket.socket() timeout functions"""
+
+    def setUp(self):
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+
     def testRecvfromTimeout(self):
         # Test recvfrom() timeout
         _timeout = 2
-        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.sock.settimeout(_timeout)
         # Prevent "Address already in use" socket exceptions
         support.bind_port(self.sock, self.localhost)
@@ -178,25 +205,14 @@ class TimeoutTestCase(unittest.TestCase):
                      "timeout (%g) is %g seconds more than expected (%g)"
                      %(_delta, self.fuzz, _timeout))
 
-    def testSend(self):
-        # Test send() timeout
-        # couldn't figure out how to test it
-        pass
-
-    def testSendto(self):
-        # Test sendto() timeout
-        # couldn't figure out how to test it
-        pass
-
-    def testSendall(self):
-        # Test sendall() timeout
-        # couldn't figure out how to test it
-        pass
-
 
 def test_main():
     support.requires('network')
-    support.run_unittest(CreationTestCase, TimeoutTestCase)
+    support.run_unittest(
+        CreationTestCase,
+        TCPTimeoutTestCase,
+        UDPTimeoutTestCase,
+    )
 
 if __name__ == "__main__":
     test_main()