1 """Unit tests for socket timeout feature."""
2
3 import functools
4 import unittest
5 from test import support
6 from test.support import socket_helper
7
8 import time
9 import errno
10 import socket
11
12
13 @functools.lru_cache()
14 def resolve_address(host, port):
15 """Resolve an (host, port) to an address.
16
17 We must perform name resolution before timeout tests, otherwise it will be
18 performed by connect().
19 """
20 with socket_helper.transient_internet(host):
21 return socket.getaddrinfo(host, port, socket.AF_INET,
22 socket.SOCK_STREAM)[0][4]
23
24
25 class ESC[4;38;5;81mCreationTestCase(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
26 """Test case for socket.gettimeout() and socket.settimeout()"""
27
28 def setUp(self):
29 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
30
31 def tearDown(self):
32 self.sock.close()
33
34 def testObjectCreation(self):
35 # Test Socket creation
36 self.assertEqual(self.sock.gettimeout(), None,
37 "timeout not disabled by default")
38
39 def testFloatReturnValue(self):
40 # Test return value of gettimeout()
41 self.sock.settimeout(7.345)
42 self.assertEqual(self.sock.gettimeout(), 7.345)
43
44 self.sock.settimeout(3)
45 self.assertEqual(self.sock.gettimeout(), 3)
46
47 self.sock.settimeout(None)
48 self.assertEqual(self.sock.gettimeout(), None)
49
50 def testReturnType(self):
51 # Test return type of gettimeout()
52 self.sock.settimeout(1)
53 self.assertEqual(type(self.sock.gettimeout()), type(1.0))
54
55 self.sock.settimeout(3.9)
56 self.assertEqual(type(self.sock.gettimeout()), type(1.0))
57
58 def testTypeCheck(self):
59 # Test type checking by settimeout()
60 self.sock.settimeout(0)
61 self.sock.settimeout(0)
62 self.sock.settimeout(0.0)
63 self.sock.settimeout(None)
64 self.assertRaises(TypeError, self.sock.settimeout, "")
65 self.assertRaises(TypeError, self.sock.settimeout, "")
66 self.assertRaises(TypeError, self.sock.settimeout, ())
67 self.assertRaises(TypeError, self.sock.settimeout, [])
68 self.assertRaises(TypeError, self.sock.settimeout, {})
69 self.assertRaises(TypeError, self.sock.settimeout, 0j)
70
71 def testRangeCheck(self):
72 # Test range checking by settimeout()
73 self.assertRaises(ValueError, self.sock.settimeout, -1)
74 self.assertRaises(ValueError, self.sock.settimeout, -1)
75 self.assertRaises(ValueError, self.sock.settimeout, -1.0)
76
77 def testTimeoutThenBlocking(self):
78 # Test settimeout() followed by setblocking()
79 self.sock.settimeout(10)
80 self.sock.setblocking(True)
81 self.assertEqual(self.sock.gettimeout(), None)
82 self.sock.setblocking(False)
83 self.assertEqual(self.sock.gettimeout(), 0.0)
84
85 self.sock.settimeout(10)
86 self.sock.setblocking(False)
87 self.assertEqual(self.sock.gettimeout(), 0.0)
88 self.sock.setblocking(True)
89 self.assertEqual(self.sock.gettimeout(), None)
90
91 def testBlockingThenTimeout(self):
92 # Test setblocking() followed by settimeout()
93 self.sock.setblocking(False)
94 self.sock.settimeout(1)
95 self.assertEqual(self.sock.gettimeout(), 1)
96
97 self.sock.setblocking(True)
98 self.sock.settimeout(1)
99 self.assertEqual(self.sock.gettimeout(), 1)
100
101
102 class ESC[4;38;5;81mTimeoutTestCase(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
103 # There are a number of tests here trying to make sure that an operation
104 # doesn't take too much longer than expected. But competing machine
105 # activity makes it inevitable that such tests will fail at times.
106 # When fuzz was at 1.0, I (tim) routinely saw bogus failures on Win2K
107 # and Win98SE. Boosting it to 2.0 helped a lot, but isn't a real
108 # solution.
109 fuzz = 2.0
110
111 localhost = socket_helper.HOST
112
113 def setUp(self):
114 raise NotImplementedError()
115
116 tearDown = setUp
117
118 def _sock_operation(self, count, timeout, method, *args):
119 """
120 Test the specified socket method.
121
122 The method is run at most `count` times and must raise a TimeoutError
123 within `timeout` + self.fuzz seconds.
124 """
125 self.sock.settimeout(timeout)
126 method = getattr(self.sock, method)
127 for i in range(count):
128 t1 = time.monotonic()
129 try:
130 method(*args)
131 except TimeoutError as e:
132 delta = time.monotonic() - t1
133 break
134 else:
135 self.fail('TimeoutError was not raised')
136 # These checks should account for timing unprecision
137 self.assertLess(delta, timeout + self.fuzz)
138 self.assertGreater(delta, timeout - 1.0)
139
140
141 class ESC[4;38;5;81mTCPTimeoutTestCase(ESC[4;38;5;149mTimeoutTestCase):
142 """TCP test case for socket.socket() timeout functions"""
143
144 def setUp(self):
145 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
146 self.addr_remote = resolve_address('www.python.org.', 80)
147
148 def tearDown(self):
149 self.sock.close()
150
151 def testConnectTimeout(self):
152 # Testing connect timeout is tricky: we need to have IP connectivity
153 # to a host that silently drops our packets. We can't simulate this
154 # from Python because it's a function of the underlying TCP/IP stack.
155 # So, the following port on the pythontest.net host has been defined:
156 blackhole = resolve_address('pythontest.net', 56666)
157
158 # Blackhole has been configured to silently drop any incoming packets.
159 # No RSTs (for TCP) or ICMP UNREACH (for UDP/ICMP) will be sent back
160 # to hosts that attempt to connect to this address: which is exactly
161 # what we need to confidently test connect timeout.
162
163 # However, we want to prevent false positives. It's not unreasonable
164 # to expect certain hosts may not be able to reach the blackhole, due
165 # to firewalling or general network configuration. In order to improve
166 # our confidence in testing the blackhole, a corresponding 'whitehole'
167 # has also been set up using one port higher:
168 whitehole = resolve_address('pythontest.net', 56667)
169
170 # This address has been configured to immediately drop any incoming
171 # packets as well, but it does it respectfully with regards to the
172 # incoming protocol. RSTs are sent for TCP packets, and ICMP UNREACH
173 # is sent for UDP/ICMP packets. This means our attempts to connect to
174 # it should be met immediately with ECONNREFUSED. The test case has
175 # been structured around this premise: if we get an ECONNREFUSED from
176 # the whitehole, we proceed with testing connect timeout against the
177 # blackhole. If we don't, we skip the test (with a message about not
178 # getting the required RST from the whitehole within the required
179 # timeframe).
180
181 # For the records, the whitehole/blackhole configuration has been set
182 # up using the 'iptables' firewall, using the following rules:
183 #
184 # -A INPUT -p tcp --destination-port 56666 -j DROP
185 # -A INPUT -p udp --destination-port 56666 -j DROP
186 # -A INPUT -p tcp --destination-port 56667 -j REJECT
187 # -A INPUT -p udp --destination-port 56667 -j REJECT
188 #
189 # See https://github.com/python/psf-salt/blob/main/pillar/base/firewall/snakebite.sls
190 # for the current configuration.
191
192 skip = True
193 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
194 timeout = support.LOOPBACK_TIMEOUT
195 sock.settimeout(timeout)
196 try:
197 sock.connect((whitehole))
198 except TimeoutError:
199 pass
200 except OSError as err:
201 if err.errno == errno.ECONNREFUSED:
202 skip = False
203 finally:
204 sock.close()
205 del sock
206
207 if skip:
208 self.skipTest(
209 "We didn't receive a connection reset (RST) packet from "
210 "{}:{} within {} seconds, so we're unable to test connect "
211 "timeout against the corresponding {}:{} (which is "
212 "configured to silently drop packets)."
213 .format(
214 whitehole[0],
215 whitehole[1],
216 timeout,
217 blackhole[0],
218 blackhole[1],
219 )
220 )
221
222 # All that hard work just to test if connect times out in 0.001s ;-)
223 self.addr_remote = blackhole
224 with socket_helper.transient_internet(self.addr_remote[0]):
225 self._sock_operation(1, 0.001, 'connect', self.addr_remote)
226
227 def testRecvTimeout(self):
228 # Test recv() timeout
229 with socket_helper.transient_internet(self.addr_remote[0]):
230 self.sock.connect(self.addr_remote)
231 self._sock_operation(1, 1.5, 'recv', 1024)
232
233 def testAcceptTimeout(self):
234 # Test accept() timeout
235 socket_helper.bind_port(self.sock, self.localhost)
236 self.sock.listen()
237 self._sock_operation(1, 1.5, 'accept')
238
239 def testSend(self):
240 # Test send() timeout
241 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
242 socket_helper.bind_port(serv, self.localhost)
243 serv.listen()
244 self.sock.connect(serv.getsockname())
245 # Send a lot of data in order to bypass buffering in the TCP stack.
246 self._sock_operation(100, 1.5, 'send', b"X" * 200000)
247
248 def testSendto(self):
249 # Test sendto() timeout
250 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
251 socket_helper.bind_port(serv, self.localhost)
252 serv.listen()
253 self.sock.connect(serv.getsockname())
254 # The address argument is ignored since we already connected.
255 self._sock_operation(100, 1.5, 'sendto', b"X" * 200000,
256 serv.getsockname())
257
258 def testSendall(self):
259 # Test sendall() timeout
260 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
261 socket_helper.bind_port(serv, self.localhost)
262 serv.listen()
263 self.sock.connect(serv.getsockname())
264 # Send a lot of data in order to bypass buffering in the TCP stack.
265 self._sock_operation(100, 1.5, 'sendall', b"X" * 200000)
266
267
268 class ESC[4;38;5;81mUDPTimeoutTestCase(ESC[4;38;5;149mTimeoutTestCase):
269 """UDP test case for socket.socket() timeout functions"""
270
271 def setUp(self):
272 self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
273
274 def tearDown(self):
275 self.sock.close()
276
277 def testRecvfromTimeout(self):
278 # Test recvfrom() timeout
279 # Prevent "Address already in use" socket exceptions
280 socket_helper.bind_port(self.sock, self.localhost)
281 self._sock_operation(1, 1.5, 'recvfrom', 1024)
282
283
284 def setUpModule():
285 support.requires('network')
286 support.requires_working_socket(module=True)
287
288
289 if __name__ == "__main__":
290 unittest.main()