|
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.addr_local = ('127.0.0.1', 25339) |
|
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.failUnlessRaises(socket.error, self.sock.connect, addr) |
|
123 _t2 = time.time() |
|
124 |
|
125 _delta = abs(_t1 - _t2) |
|
126 self.assert_(_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 self.sock.connect(self.addr_remote) |
|
134 self.sock.settimeout(_timeout) |
|
135 |
|
136 _t1 = time.time() |
|
137 self.failUnlessRaises(socket.error, self.sock.recv, 1024) |
|
138 _t2 = time.time() |
|
139 |
|
140 _delta = abs(_t1 - _t2) |
|
141 self.assert_(_delta < _timeout + self.fuzz, |
|
142 "timeout (%g) is %g seconds more than expected (%g)" |
|
143 %(_delta, self.fuzz, _timeout)) |
|
144 |
|
145 def testAcceptTimeout(self): |
|
146 # Test accept() timeout |
|
147 _timeout = 2 |
|
148 self.sock.settimeout(_timeout) |
|
149 self.sock.bind(self.addr_local) |
|
150 self.sock.listen(5) |
|
151 |
|
152 _t1 = time.time() |
|
153 self.failUnlessRaises(socket.error, self.sock.accept) |
|
154 _t2 = time.time() |
|
155 |
|
156 _delta = abs(_t1 - _t2) |
|
157 self.assert_(_delta < _timeout + self.fuzz, |
|
158 "timeout (%g) is %g seconds more than expected (%g)" |
|
159 %(_delta, self.fuzz, _timeout)) |
|
160 |
|
161 def testRecvfromTimeout(self): |
|
162 # Test recvfrom() timeout |
|
163 _timeout = 2 |
|
164 self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
|
165 self.sock.settimeout(_timeout) |
|
166 self.sock.bind(self.addr_local) |
|
167 |
|
168 _t1 = time.time() |
|
169 self.failUnlessRaises(socket.error, self.sock.recvfrom, 8192) |
|
170 _t2 = time.time() |
|
171 |
|
172 _delta = abs(_t1 - _t2) |
|
173 self.assert_(_delta < _timeout + self.fuzz, |
|
174 "timeout (%g) is %g seconds more than expected (%g)" |
|
175 %(_delta, self.fuzz, _timeout)) |
|
176 |
|
177 def testSend(self): |
|
178 # Test send() timeout |
|
179 # couldn't figure out how to test it |
|
180 pass |
|
181 |
|
182 def testSendto(self): |
|
183 # Test sendto() timeout |
|
184 # couldn't figure out how to test it |
|
185 pass |
|
186 |
|
187 def testSendall(self): |
|
188 # Test sendall() timeout |
|
189 # couldn't figure out how to test it |
|
190 pass |
|
191 |
|
192 |
|
193 def test_main(): |
|
194 test_support.requires('network') |
|
195 test_support.run_unittest(CreationTestCase, TimeoutTestCase) |
|
196 |
|
197 if __name__ == "__main__": |
|
198 test_main() |