Commit | Line | Data |
---|---|---|
920dae64 AT |
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 | # Test connect() timeout | |
111 | _timeout = 0.001 | |
112 | self.sock.settimeout(_timeout) | |
113 | ||
114 | _t1 = time.time() | |
115 | self.failUnlessRaises(socket.error, self.sock.connect, | |
116 | self.addr_remote) | |
117 | _t2 = time.time() | |
118 | ||
119 | _delta = abs(_t1 - _t2) | |
120 | self.assert_(_delta < _timeout + self.fuzz, | |
121 | "timeout (%g) is more than %g seconds more than expected (%g)" | |
122 | %(_delta, self.fuzz, _timeout)) | |
123 | ||
124 | def testRecvTimeout(self): | |
125 | # Test recv() timeout | |
126 | _timeout = 0.02 | |
127 | self.sock.connect(self.addr_remote) | |
128 | self.sock.settimeout(_timeout) | |
129 | ||
130 | _t1 = time.time() | |
131 | self.failUnlessRaises(socket.error, self.sock.recv, 1024) | |
132 | _t2 = time.time() | |
133 | ||
134 | _delta = abs(_t1 - _t2) | |
135 | self.assert_(_delta < _timeout + self.fuzz, | |
136 | "timeout (%g) is %g seconds more than expected (%g)" | |
137 | %(_delta, self.fuzz, _timeout)) | |
138 | ||
139 | def testAcceptTimeout(self): | |
140 | # Test accept() timeout | |
141 | _timeout = 2 | |
142 | self.sock.settimeout(_timeout) | |
143 | self.sock.bind(self.addr_local) | |
144 | self.sock.listen(5) | |
145 | ||
146 | _t1 = time.time() | |
147 | self.failUnlessRaises(socket.error, self.sock.accept) | |
148 | _t2 = time.time() | |
149 | ||
150 | _delta = abs(_t1 - _t2) | |
151 | self.assert_(_delta < _timeout + self.fuzz, | |
152 | "timeout (%g) is %g seconds more than expected (%g)" | |
153 | %(_delta, self.fuzz, _timeout)) | |
154 | ||
155 | def testRecvfromTimeout(self): | |
156 | # Test recvfrom() timeout | |
157 | _timeout = 2 | |
158 | self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) | |
159 | self.sock.settimeout(_timeout) | |
160 | self.sock.bind(self.addr_local) | |
161 | ||
162 | _t1 = time.time() | |
163 | self.failUnlessRaises(socket.error, self.sock.recvfrom, 8192) | |
164 | _t2 = time.time() | |
165 | ||
166 | _delta = abs(_t1 - _t2) | |
167 | self.assert_(_delta < _timeout + self.fuzz, | |
168 | "timeout (%g) is %g seconds more than expected (%g)" | |
169 | %(_delta, self.fuzz, _timeout)) | |
170 | ||
171 | def testSend(self): | |
172 | # Test send() timeout | |
173 | # couldn't figure out how to test it | |
174 | pass | |
175 | ||
176 | def testSendto(self): | |
177 | # Test sendto() timeout | |
178 | # couldn't figure out how to test it | |
179 | pass | |
180 | ||
181 | def testSendall(self): | |
182 | # Test sendall() timeout | |
183 | # couldn't figure out how to test it | |
184 | pass | |
185 | ||
186 | ||
187 | def test_main(): | |
188 | test_support.requires('network') | |
189 | test_support.run_unittest(CreationTestCase, TimeoutTestCase) | |
190 | ||
191 | if __name__ == "__main__": | |
192 | test_main() |