Python code coverage for Lib/test/test_timeout.py

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