1 """Unit tests for socket timeout feature.""" 2 3 import unittest 4 from test import test_support 5 6 # This requires the 'network' resource as given on the regrtest command line. 7 skip_expected = not test_support.is_resource_enabled('network') 8 9 import time 10 import socket 11 12 13 class CreationTestCase(unittest.TestCase): 14 """Test case for socket.gettimeout() and socket.settimeout()""" 15 16 def setUp(self): 17 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 18 19 def tearDown(self): 20 self.sock.close() 21 22 def testObjectCreation(self): 23 # Test Socket creation 24 self.assertEqual(self.sock.gettimeout(), None, 25 "timeout not disabled by default") 26 27 def testFloatReturnValue(self): 28 # Test return value of gettimeout() 29 self.sock.settimeout(7.345) 30 self.assertEqual(self.sock.gettimeout(), 7.345) 31 32 self.sock.settimeout(3) 33 self.assertEqual(self.sock.gettimeout(), 3) 34 35 self.sock.settimeout(None) 36 self.assertEqual(self.sock.gettimeout(), None) 37 38 def testReturnType(self): 39 # Test return type of gettimeout() 40 self.sock.settimeout(1) 41 self.assertEqual(type(self.sock.gettimeout()), type(1.0)) 42 43 self.sock.settimeout(3.9) 44 self.assertEqual(type(self.sock.gettimeout()), type(1.0)) 45 46 def testTypeCheck(self): 47 # Test type checking by settimeout() 48 self.sock.settimeout(0) 49 self.sock.settimeout(0L) 50 self.sock.settimeout(0.0) 51 self.sock.settimeout(None) 52 self.assertRaises(TypeError, self.sock.settimeout, "") 53 self.assertRaises(TypeError, self.sock.settimeout, u"") 54 self.assertRaises(TypeError, self.sock.settimeout, ()) 55 self.assertRaises(TypeError, self.sock.settimeout, []) 56 self.assertRaises(TypeError, self.sock.settimeout, {}) 57 self.assertRaises(TypeError, self.sock.settimeout, 0j) 58 59 def testRangeCheck(self): 60 # Test range checking by settimeout() 61 self.assertRaises(ValueError, self.sock.settimeout, -1) 62 self.assertRaises(ValueError, self.sock.settimeout, -1L) 63 self.assertRaises(ValueError, self.sock.settimeout, -1.0) 64 65 def testTimeoutThenBlocking(self): 66 # Test settimeout() followed by setblocking() 67 self.sock.settimeout(10) 68 self.sock.setblocking(1) 69 self.assertEqual(self.sock.gettimeout(), None) 70 self.sock.setblocking(0) 71 self.assertEqual(self.sock.gettimeout(), 0.0) 72 73 self.sock.settimeout(10) 74 self.sock.setblocking(0) 75 self.assertEqual(self.sock.gettimeout(), 0.0) 76 self.sock.setblocking(1) 77 self.assertEqual(self.sock.gettimeout(), None) 78 79 def testBlockingThenTimeout(self): 80 # Test setblocking() followed by settimeout() 81 self.sock.setblocking(0) 82 self.sock.settimeout(1) 83 self.assertEqual(self.sock.gettimeout(), 1) 84 85 self.sock.setblocking(1) 86 self.sock.settimeout(1) 87 self.assertEqual(self.sock.gettimeout(), 1) 88 89 90 class TimeoutTestCase(unittest.TestCase): 91 """Test case for socket.socket() timeout functions""" 92 93 # There are a number of tests here trying to make sure that an operation 94 # doesn't take too much longer than expected. But competing machine 95 # activity makes it inevitable that such tests will fail at times. 96 # When fuzz was at 1.0, I (tim) routinely saw bogus failures on Win2K 97 # and Win98SE. Boosting it to 2.0 helped a lot, but isn't a real 98 # solution. 99 fuzz = 2.0 100 101 def setUp(self): 102 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 103 self.addr_remote = ('www.python.org.', 80) 104 self.localhost = '127.0.0.1' 105 106 def tearDown(self): 107 self.sock.close() 108 109 def testConnectTimeout(self): 110 # Choose a private address that is unlikely to exist to prevent 111 # failures due to the connect succeeding before the timeout. 112 # Use a dotted IP address to avoid including the DNS lookup time 113 # with the connect time. This avoids failing the assertion that 114 # the timeout occurred fast enough. 115 addr = ('10.0.0.0', 12345) 116 117 # Test connect() timeout 118 _timeout = 0.001 119 self.sock.settimeout(_timeout) 120 121 _t1 = time.time() 122 self.assertRaises(socket.error, self.sock.connect, addr) 123 _t2 = time.time() 124 125 _delta = abs(_t1 - _t2) 126 self.assertTrue(_delta < _timeout + self.fuzz, 127 "timeout (%g) is more than %g seconds more than expected (%g)" 128 %(_delta, self.fuzz, _timeout)) 129 130 def testRecvTimeout(self): 131 # Test recv() timeout 132 _timeout = 0.02 133 134 with test_support.transient_internet(self.addr_remote[0]): 135 self.sock.connect(self.addr_remote) 136 self.sock.settimeout(_timeout) 137 138 _t1 = time.time() 139 self.assertRaises(socket.timeout, self.sock.recv, 1024) 140 _t2 = time.time() 141 142 _delta = abs(_t1 - _t2) 143 self.assertTrue(_delta < _timeout + self.fuzz, 144 "timeout (%g) is %g seconds more than expected (%g)" 145 %(_delta, self.fuzz, _timeout)) 146 147 def testAcceptTimeout(self): 148 # Test accept() timeout 149 _timeout = 2 150 self.sock.settimeout(_timeout) 151 # Prevent "Address already in use" socket exceptions 152 test_support.bind_port(self.sock, self.localhost) 153 self.sock.listen(5) 154 155 _t1 = time.time() 156 self.assertRaises(socket.error, self.sock.accept) 157 _t2 = time.time() 158 159 _delta = abs(_t1 - _t2) 160 self.assertTrue(_delta < _timeout + self.fuzz, 161 "timeout (%g) is %g seconds more than expected (%g)" 162 %(_delta, self.fuzz, _timeout)) 163 164 def testRecvfromTimeout(self): 165 # Test recvfrom() timeout 166 _timeout = 2 167 self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 168 self.sock.settimeout(_timeout) 169 # Prevent "Address already in use" socket exceptions 170 test_support.bind_port(self.sock, self.localhost) 171 172 _t1 = time.time() 173 self.assertRaises(socket.error, self.sock.recvfrom, 8192) 174 _t2 = time.time() 175 176 _delta = abs(_t1 - _t2) 177 self.assertTrue(_delta < _timeout + self.fuzz, 178 "timeout (%g) is %g seconds more than expected (%g)" 179 %(_delta, self.fuzz, _timeout)) 180 181 @unittest.skip('test not implemented') 182 def testSend(self): 183 # Test send() timeout 184 # couldn't figure out how to test it 185 pass 186 187 @unittest.skip('test not implemented') 188 def testSendto(self): 189 # Test sendto() timeout 190 # couldn't figure out how to test it 191 pass 192 193 @unittest.skip('test not implemented') 194 def testSendall(self): 195 # Test sendall() timeout 196 # couldn't figure out how to test it 197 pass 198 199 200 def test_main(): 201 test_support.requires('network') 202 test_support.run_unittest(CreationTestCase, TimeoutTestCase) 203 204 if __name__ == "__main__": 205 test_main() 206
