Python code coverage for Lib/test/test_socketserver.py

#	count	content
1	n/a	"""
2	n/a	Test suite for socketserver.
3	n/a	"""
4	n/a
5	n/a	import contextlib
6	n/a	import io
7	n/a	import os
8	n/a	import select
9	n/a	import signal
10	n/a	import socket
11	n/a	import tempfile
12	n/a	import unittest
13	n/a	import socketserver
14	n/a
15	n/a	import test.support
16	n/a	from test.support import reap_children, reap_threads, verbose
17	n/a	try:
18	n/a	import threading
19	n/a	except ImportError:
20	n/a	threading = None
21	n/a
22	n/a	test.support.requires("network")
23	n/a
24	n/a	TEST_STR = b"hello world\n"
25	n/a	HOST = test.support.HOST
26	n/a
27	n/a	HAVE_UNIX_SOCKETS = hasattr(socket, "AF_UNIX")
28	n/a	requires_unix_sockets = unittest.skipUnless(HAVE_UNIX_SOCKETS,
29	n/a	'requires Unix sockets')
30	n/a	HAVE_FORKING = hasattr(os, "fork")
31	n/a	requires_forking = unittest.skipUnless(HAVE_FORKING, 'requires forking')
32	n/a
33	n/a	def signal_alarm(n):
34	n/a	"""Call signal.alarm when it exists (i.e. not on Windows)."""
35	n/a	if hasattr(signal, 'alarm'):
36	n/a	signal.alarm(n)
37	n/a
38	n/a	# Remember real select() to avoid interferences with mocking
39	n/a	_real_select = select.select
40	n/a
41	n/a	def receive(sock, n, timeout=20):
42	n/a	r, w, x = _real_select([sock], [], [], timeout)
43	n/a	if sock in r:
44	n/a	return sock.recv(n)
45	n/a	else:
46	n/a	raise RuntimeError("timed out on %r" % (sock,))
47	n/a
48	n/a	if HAVE_UNIX_SOCKETS and HAVE_FORKING:
49	n/a	class ForkingUnixStreamServer(socketserver.ForkingMixIn,
50	n/a	socketserver.UnixStreamServer):
51	n/a	pass
52	n/a
53	n/a	class ForkingUnixDatagramServer(socketserver.ForkingMixIn,
54	n/a	socketserver.UnixDatagramServer):
55	n/a	pass
56	n/a
57	n/a
58	n/a	@contextlib.contextmanager
59	n/a	def simple_subprocess(testcase):
60	n/a	"""Tests that a custom child process is not waited on (Issue 1540386)"""
61	n/a	pid = os.fork()
62	n/a	if pid == 0:
63	n/a	# Don't raise an exception; it would be caught by the test harness.
64	n/a	os._exit(72)
65	n/a	yield None
66	n/a	pid2, status = os.waitpid(pid, 0)
67	n/a	testcase.assertEqual(pid2, pid)
68	n/a	testcase.assertEqual(72 << 8, status)
69	n/a
70	n/a
71	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
72	n/a	class SocketServerTest(unittest.TestCase):
73	n/a	"""Test all socket servers."""
74	n/a
75	n/a	def setUp(self):
76	n/a	signal_alarm(60) # Kill deadlocks after 60 seconds.
77	n/a	self.port_seed = 0
78	n/a	self.test_files = []
79	n/a
80	n/a	def tearDown(self):
81	n/a	signal_alarm(0) # Didn't deadlock.
82	n/a	reap_children()
83	n/a
84	n/a	for fn in self.test_files:
85	n/a	try:
86	n/a	os.remove(fn)
87	n/a	except OSError:
88	n/a	pass
89	n/a	self.test_files[:] = []
90	n/a
91	n/a	def pickaddr(self, proto):
92	n/a	if proto == socket.AF_INET:
93	n/a	return (HOST, 0)
94	n/a	else:
95	n/a	# XXX: We need a way to tell AF_UNIX to pick its own name
96	n/a	# like AF_INET provides port==0.
97	n/a	dir = None
98	n/a	fn = tempfile.mktemp(prefix='unix_socket.', dir=dir)
99	n/a	self.test_files.append(fn)
100	n/a	return fn
101	n/a
102	n/a	def make_server(self, addr, svrcls, hdlrbase):
103	n/a	class MyServer(svrcls):
104	n/a	def handle_error(self, request, client_address):
105	n/a	self.close_request(request)
106	n/a	raise
107	n/a
108	n/a	class MyHandler(hdlrbase):
109	n/a	def handle(self):
110	n/a	line = self.rfile.readline()
111	n/a	self.wfile.write(line)
112	n/a
113	n/a	if verbose: print("creating server")
114	n/a	server = MyServer(addr, MyHandler)
115	n/a	self.assertEqual(server.server_address, server.socket.getsockname())
116	n/a	return server
117	n/a
118	n/a	@reap_threads
119	n/a	def run_server(self, svrcls, hdlrbase, testfunc):
120	n/a	server = self.make_server(self.pickaddr(svrcls.address_family),
121	n/a	svrcls, hdlrbase)
122	n/a	# We had the OS pick a port, so pull the real address out of
123	n/a	# the server.
124	n/a	addr = server.server_address
125	n/a	if verbose:
126	n/a	print("ADDR =", addr)
127	n/a	print("CLASS =", svrcls)
128	n/a
129	n/a	t = threading.Thread(
130	n/a	name='%s serving' % svrcls,
131	n/a	target=server.serve_forever,
132	n/a	# Short poll interval to make the test finish quickly.
133	n/a	# Time between requests is short enough that we won't wake
134	n/a	# up spuriously too many times.
135	n/a	kwargs={'poll_interval':0.01})
136	n/a	t.daemon = True # In case this function raises.
137	n/a	t.start()
138	n/a	if verbose: print("server running")
139	n/a	for i in range(3):
140	n/a	if verbose: print("test client", i)
141	n/a	testfunc(svrcls.address_family, addr)
142	n/a	if verbose: print("waiting for server")
143	n/a	server.shutdown()
144	n/a	t.join()
145	n/a	server.server_close()
146	n/a	self.assertEqual(-1, server.socket.fileno())
147	n/a	if verbose: print("done")
148	n/a
149	n/a	def stream_examine(self, proto, addr):
150	n/a	s = socket.socket(proto, socket.SOCK_STREAM)
151	n/a	s.connect(addr)
152	n/a	s.sendall(TEST_STR)
153	n/a	buf = data = receive(s, 100)
154	n/a	while data and b'\n' not in buf:
155	n/a	data = receive(s, 100)
156	n/a	buf += data
157	n/a	self.assertEqual(buf, TEST_STR)
158	n/a	s.close()
159	n/a
160	n/a	def dgram_examine(self, proto, addr):
161	n/a	s = socket.socket(proto, socket.SOCK_DGRAM)
162	n/a	if HAVE_UNIX_SOCKETS and proto == socket.AF_UNIX:
163	n/a	s.bind(self.pickaddr(proto))
164	n/a	s.sendto(TEST_STR, addr)
165	n/a	buf = data = receive(s, 100)
166	n/a	while data and b'\n' not in buf:
167	n/a	data = receive(s, 100)
168	n/a	buf += data
169	n/a	self.assertEqual(buf, TEST_STR)
170	n/a	s.close()
171	n/a
172	n/a	def test_TCPServer(self):
173	n/a	self.run_server(socketserver.TCPServer,
174	n/a	socketserver.StreamRequestHandler,
175	n/a	self.stream_examine)
176	n/a
177	n/a	def test_ThreadingTCPServer(self):
178	n/a	self.run_server(socketserver.ThreadingTCPServer,
179	n/a	socketserver.StreamRequestHandler,
180	n/a	self.stream_examine)
181	n/a
182	n/a	@requires_forking
183	n/a	def test_ForkingTCPServer(self):
184	n/a	with simple_subprocess(self):
185	n/a	self.run_server(socketserver.ForkingTCPServer,
186	n/a	socketserver.StreamRequestHandler,
187	n/a	self.stream_examine)
188	n/a
189	n/a	@requires_unix_sockets
190	n/a	def test_UnixStreamServer(self):
191	n/a	self.run_server(socketserver.UnixStreamServer,
192	n/a	socketserver.StreamRequestHandler,
193	n/a	self.stream_examine)
194	n/a
195	n/a	@requires_unix_sockets
196	n/a	def test_ThreadingUnixStreamServer(self):
197	n/a	self.run_server(socketserver.ThreadingUnixStreamServer,
198	n/a	socketserver.StreamRequestHandler,
199	n/a	self.stream_examine)
200	n/a
201	n/a	@requires_unix_sockets
202	n/a	@requires_forking
203	n/a	def test_ForkingUnixStreamServer(self):
204	n/a	with simple_subprocess(self):
205	n/a	self.run_server(ForkingUnixStreamServer,
206	n/a	socketserver.StreamRequestHandler,
207	n/a	self.stream_examine)
208	n/a
209	n/a	def test_UDPServer(self):
210	n/a	self.run_server(socketserver.UDPServer,
211	n/a	socketserver.DatagramRequestHandler,
212	n/a	self.dgram_examine)
213	n/a
214	n/a	def test_ThreadingUDPServer(self):
215	n/a	self.run_server(socketserver.ThreadingUDPServer,
216	n/a	socketserver.DatagramRequestHandler,
217	n/a	self.dgram_examine)
218	n/a
219	n/a	@requires_forking
220	n/a	def test_ForkingUDPServer(self):
221	n/a	with simple_subprocess(self):
222	n/a	self.run_server(socketserver.ForkingUDPServer,
223	n/a	socketserver.DatagramRequestHandler,
224	n/a	self.dgram_examine)
225	n/a
226	n/a	@requires_unix_sockets
227	n/a	def test_UnixDatagramServer(self):
228	n/a	self.run_server(socketserver.UnixDatagramServer,
229	n/a	socketserver.DatagramRequestHandler,
230	n/a	self.dgram_examine)
231	n/a
232	n/a	@requires_unix_sockets
233	n/a	def test_ThreadingUnixDatagramServer(self):
234	n/a	self.run_server(socketserver.ThreadingUnixDatagramServer,
235	n/a	socketserver.DatagramRequestHandler,
236	n/a	self.dgram_examine)
237	n/a
238	n/a	@requires_unix_sockets
239	n/a	@requires_forking
240	n/a	def test_ForkingUnixDatagramServer(self):
241	n/a	self.run_server(ForkingUnixDatagramServer,
242	n/a	socketserver.DatagramRequestHandler,
243	n/a	self.dgram_examine)
244	n/a
245	n/a	@reap_threads
246	n/a	def test_shutdown(self):
247	n/a	# Issue #2302: shutdown() should always succeed in making an
248	n/a	# other thread leave serve_forever().
249	n/a	class MyServer(socketserver.TCPServer):
250	n/a	pass
251	n/a
252	n/a	class MyHandler(socketserver.StreamRequestHandler):
253	n/a	pass
254	n/a
255	n/a	threads = []
256	n/a	for i in range(20):
257	n/a	s = MyServer((HOST, 0), MyHandler)
258	n/a	t = threading.Thread(
259	n/a	name='MyServer serving',
260	n/a	target=s.serve_forever,
261	n/a	kwargs={'poll_interval':0.01})
262	n/a	t.daemon = True # In case this function raises.
263	n/a	threads.append((t, s))
264	n/a	for t, s in threads:
265	n/a	t.start()
266	n/a	s.shutdown()
267	n/a	for t, s in threads:
268	n/a	t.join()
269	n/a	s.server_close()
270	n/a
271	n/a	def test_tcpserver_bind_leak(self):
272	n/a	# Issue #22435: the server socket wouldn't be closed if bind()/listen()
273	n/a	# failed.
274	n/a	# Create many servers for which bind() will fail, to see if this result
275	n/a	# in FD exhaustion.
276	n/a	for i in range(1024):
277	n/a	with self.assertRaises(OverflowError):
278	n/a	socketserver.TCPServer((HOST, -1),
279	n/a	socketserver.StreamRequestHandler)
280	n/a
281	n/a	def test_context_manager(self):
282	n/a	with socketserver.TCPServer((HOST, 0),
283	n/a	socketserver.StreamRequestHandler) as server:
284	n/a	pass
285	n/a	self.assertEqual(-1, server.socket.fileno())
286	n/a
287	n/a
288	n/a	class ErrorHandlerTest(unittest.TestCase):
289	n/a	"""Test that the servers pass normal exceptions from the handler to
290	n/a	handle_error(), and that exiting exceptions like SystemExit and
291	n/a	KeyboardInterrupt are not passed."""
292	n/a
293	n/a	def tearDown(self):
294	n/a	test.support.unlink(test.support.TESTFN)
295	n/a
296	n/a	def test_sync_handled(self):
297	n/a	BaseErrorTestServer(ValueError)
298	n/a	self.check_result(handled=True)
299	n/a
300	n/a	def test_sync_not_handled(self):
301	n/a	with self.assertRaises(SystemExit):
302	n/a	BaseErrorTestServer(SystemExit)
303	n/a	self.check_result(handled=False)
304	n/a
305	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
306	n/a	def test_threading_handled(self):
307	n/a	ThreadingErrorTestServer(ValueError)
308	n/a	self.check_result(handled=True)
309	n/a
310	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
311	n/a	def test_threading_not_handled(self):
312	n/a	ThreadingErrorTestServer(SystemExit)
313	n/a	self.check_result(handled=False)
314	n/a
315	n/a	@requires_forking
316	n/a	def test_forking_handled(self):
317	n/a	ForkingErrorTestServer(ValueError)
318	n/a	self.check_result(handled=True)
319	n/a
320	n/a	@requires_forking
321	n/a	def test_forking_not_handled(self):
322	n/a	ForkingErrorTestServer(SystemExit)
323	n/a	self.check_result(handled=False)
324	n/a
325	n/a	def check_result(self, handled):
326	n/a	with open(test.support.TESTFN) as log:
327	n/a	expected = 'Handler called\n' + 'Error handled\n' * handled
328	n/a	self.assertEqual(log.read(), expected)
329	n/a
330	n/a
331	n/a	class BaseErrorTestServer(socketserver.TCPServer):
332	n/a	def __init__(self, exception):
333	n/a	self.exception = exception
334	n/a	super().__init__((HOST, 0), BadHandler)
335	n/a	with socket.create_connection(self.server_address):
336	n/a	pass
337	n/a	try:
338	n/a	self.handle_request()
339	n/a	finally:
340	n/a	self.server_close()
341	n/a	self.wait_done()
342	n/a
343	n/a	def handle_error(self, request, client_address):
344	n/a	with open(test.support.TESTFN, 'a') as log:
345	n/a	log.write('Error handled\n')
346	n/a
347	n/a	def wait_done(self):
348	n/a	pass
349	n/a
350	n/a
351	n/a	class BadHandler(socketserver.BaseRequestHandler):
352	n/a	def handle(self):
353	n/a	with open(test.support.TESTFN, 'a') as log:
354	n/a	log.write('Handler called\n')
355	n/a	raise self.server.exception('Test error')
356	n/a
357	n/a
358	n/a	class ThreadingErrorTestServer(socketserver.ThreadingMixIn,
359	n/a	BaseErrorTestServer):
360	n/a	def __init__(self, pos, *kw):
361	n/a	self.done = threading.Event()
362	n/a	super().__init__(pos, *kw)
363	n/a
364	n/a	def shutdown_request(self, pos, *kw):
365	n/a	super().shutdown_request(pos, *kw)
366	n/a	self.done.set()
367	n/a
368	n/a	def wait_done(self):
369	n/a	self.done.wait()
370	n/a
371	n/a
372	n/a	if HAVE_FORKING:
373	n/a	class ForkingErrorTestServer(socketserver.ForkingMixIn, BaseErrorTestServer):
374	n/a	def wait_done(self):
375	n/a	[child] = self.active_children
376	n/a	os.waitpid(child, 0)
377	n/a	self.active_children.clear()
378	n/a
379	n/a
380	n/a	class SocketWriterTest(unittest.TestCase):
381	n/a	def test_basics(self):
382	n/a	class Handler(socketserver.StreamRequestHandler):
383	n/a	def handle(self):
384	n/a	self.server.wfile = self.wfile
385	n/a	self.server.wfile_fileno = self.wfile.fileno()
386	n/a	self.server.request_fileno = self.request.fileno()
387	n/a
388	n/a	server = socketserver.TCPServer((HOST, 0), Handler)
389	n/a	self.addCleanup(server.server_close)
390	n/a	s = socket.socket(
391	n/a	server.address_family, socket.SOCK_STREAM, socket.IPPROTO_TCP)
392	n/a	with s:
393	n/a	s.connect(server.server_address)
394	n/a	server.handle_request()
395	n/a	self.assertIsInstance(server.wfile, io.BufferedIOBase)
396	n/a	self.assertEqual(server.wfile_fileno, server.request_fileno)
397	n/a
398	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
399	n/a	def test_write(self):
400	n/a	# Test that wfile.write() sends data immediately, and that it does
401	n/a	# not truncate sends when interrupted by a Unix signal
402	n/a	pthread_kill = test.support.get_attribute(signal, 'pthread_kill')
403	n/a
404	n/a	class Handler(socketserver.StreamRequestHandler):
405	n/a	def handle(self):
406	n/a	self.server.sent1 = self.wfile.write(b'write data\n')
407	n/a	# Should be sent immediately, without requiring flush()
408	n/a	self.server.received = self.rfile.readline()
409	n/a	big_chunk = b'\0' * test.support.SOCK_MAX_SIZE
410	n/a	self.server.sent2 = self.wfile.write(big_chunk)
411	n/a
412	n/a	server = socketserver.TCPServer((HOST, 0), Handler)
413	n/a	self.addCleanup(server.server_close)
414	n/a	interrupted = threading.Event()
415	n/a
416	n/a	def signal_handler(signum, frame):
417	n/a	interrupted.set()
418	n/a
419	n/a	original = signal.signal(signal.SIGUSR1, signal_handler)
420	n/a	self.addCleanup(signal.signal, signal.SIGUSR1, original)
421	n/a	response1 = None
422	n/a	received2 = None
423	n/a	main_thread = threading.get_ident()
424	n/a
425	n/a	def run_client():
426	n/a	s = socket.socket(server.address_family, socket.SOCK_STREAM,
427	n/a	socket.IPPROTO_TCP)
428	n/a	with s, s.makefile('rb') as reader:
429	n/a	s.connect(server.server_address)
430	n/a	nonlocal response1
431	n/a	response1 = reader.readline()
432	n/a	s.sendall(b'client response\n')
433	n/a
434	n/a	reader.read(100)
435	n/a	# The main thread should now be blocking in a send() syscall.
436	n/a	# But in theory, it could get interrupted by other signals,
437	n/a	# and then retried. So keep sending the signal in a loop, in
438	n/a	# case an earlier signal happens to be delivered at an
439	n/a	# inconvenient moment.
440	n/a	while True:
441	n/a	pthread_kill(main_thread, signal.SIGUSR1)
442	n/a	if interrupted.wait(timeout=float(1)):
443	n/a	break
444	n/a	nonlocal received2
445	n/a	received2 = len(reader.read())
446	n/a
447	n/a	background = threading.Thread(target=run_client)
448	n/a	background.start()
449	n/a	server.handle_request()
450	n/a	background.join()
451	n/a	self.assertEqual(server.sent1, len(response1))
452	n/a	self.assertEqual(response1, b'write data\n')
453	n/a	self.assertEqual(server.received, b'client response\n')
454	n/a	self.assertEqual(server.sent2, test.support.SOCK_MAX_SIZE)
455	n/a	self.assertEqual(received2, test.support.SOCK_MAX_SIZE - 100)
456	n/a
457	n/a
458	n/a	class MiscTestCase(unittest.TestCase):
459	n/a
460	n/a	def test_all(self):
461	n/a	# objects defined in the module should be in __all__
462	n/a	expected = []
463	n/a	for name in dir(socketserver):
464	n/a	if not name.startswith('_'):
465	n/a	mod_object = getattr(socketserver, name)
466	n/a	if getattr(mod_object, '__module__', None) == 'socketserver':
467	n/a	expected.append(name)
468	n/a	self.assertCountEqual(socketserver.__all__, expected)
469	n/a
470	n/a	def test_shutdown_request_called_if_verify_request_false(self):
471	n/a	# Issue #26309: BaseServer should call shutdown_request even if
472	n/a	# verify_request is False
473	n/a
474	n/a	class MyServer(socketserver.TCPServer):
475	n/a	def verify_request(self, request, client_address):
476	n/a	return False
477	n/a
478	n/a	shutdown_called = 0
479	n/a	def shutdown_request(self, request):
480	n/a	self.shutdown_called += 1
481	n/a	socketserver.TCPServer.shutdown_request(self, request)
482	n/a
483	n/a	server = MyServer((HOST, 0), socketserver.StreamRequestHandler)
484	n/a	s = socket.socket(server.address_family, socket.SOCK_STREAM)
485	n/a	s.connect(server.server_address)
486	n/a	s.close()
487	n/a	server.handle_request()
488	n/a	self.assertEqual(server.shutdown_called, 1)
489	n/a	server.server_close()
490	n/a
491	n/a
492	n/a	if __name__ == "__main__":
493	n/a	unittest.main()