Python code coverage for Lib/test/test_signal.py

#	count	content
1	n/a	import unittest
2	n/a	from test import support
3	n/a	from contextlib import closing
4	n/a	import select
5	n/a	import signal
6	n/a	import socket
7	n/a	import struct
8	n/a	import subprocess
9	n/a	import sys, os, time, errno
10	n/a	from test.support.script_helper import assert_python_ok, spawn_python
11	n/a	try:
12	n/a	import threading
13	n/a	except ImportError:
14	n/a	threading = None
15	n/a	try:
16	n/a	import _testcapi
17	n/a	except ImportError:
18	n/a	_testcapi = None
19	n/a
20	n/a
21	n/a	class GenericTests(unittest.TestCase):
22	n/a
23	n/a	@unittest.skipIf(threading is None, "test needs threading module")
24	n/a	def test_enums(self):
25	n/a	for name in dir(signal):
26	n/a	sig = getattr(signal, name)
27	n/a	if name in {'SIG_DFL', 'SIG_IGN'}:
28	n/a	self.assertIsInstance(sig, signal.Handlers)
29	n/a	elif name in {'SIG_BLOCK', 'SIG_UNBLOCK', 'SIG_SETMASK'}:
30	n/a	self.assertIsInstance(sig, signal.Sigmasks)
31	n/a	elif name.startswith('SIG') and not name.startswith('SIG_'):
32	n/a	self.assertIsInstance(sig, signal.Signals)
33	n/a	elif name.startswith('CTRL_'):
34	n/a	self.assertIsInstance(sig, signal.Signals)
35	n/a	self.assertEqual(sys.platform, "win32")
36	n/a
37	n/a
38	n/a	@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
39	n/a	class PosixTests(unittest.TestCase):
40	n/a	def trivial_signal_handler(self, *args):
41	n/a	pass
42	n/a
43	n/a	def test_out_of_range_signal_number_raises_error(self):
44	n/a	self.assertRaises(ValueError, signal.getsignal, 4242)
45	n/a
46	n/a	self.assertRaises(ValueError, signal.signal, 4242,
47	n/a	self.trivial_signal_handler)
48	n/a
49	n/a	def test_setting_signal_handler_to_none_raises_error(self):
50	n/a	self.assertRaises(TypeError, signal.signal,
51	n/a	signal.SIGUSR1, None)
52	n/a
53	n/a	def test_getsignal(self):
54	n/a	hup = signal.signal(signal.SIGHUP, self.trivial_signal_handler)
55	n/a	self.assertIsInstance(hup, signal.Handlers)
56	n/a	self.assertEqual(signal.getsignal(signal.SIGHUP),
57	n/a	self.trivial_signal_handler)
58	n/a	signal.signal(signal.SIGHUP, hup)
59	n/a	self.assertEqual(signal.getsignal(signal.SIGHUP), hup)
60	n/a
61	n/a	# Issue 3864, unknown if this affects earlier versions of freebsd also
62	n/a	@unittest.skipIf(sys.platform=='freebsd6',
63	n/a	'inter process signals not reliable (do not mix well with threading) '
64	n/a	'on freebsd6')
65	n/a	def test_interprocess_signal(self):
66	n/a	dirname = os.path.dirname(__file__)
67	n/a	script = os.path.join(dirname, 'signalinterproctester.py')
68	n/a	assert_python_ok(script)
69	n/a
70	n/a
71	n/a	@unittest.skipUnless(sys.platform == "win32", "Windows specific")
72	n/a	class WindowsSignalTests(unittest.TestCase):
73	n/a	def test_issue9324(self):
74	n/a	# Updated for issue #10003, adding SIGBREAK
75	n/a	handler = lambda x, y: None
76	n/a	checked = set()
77	n/a	for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
78	n/a	signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
79	n/a	signal.SIGTERM):
80	n/a	# Set and then reset a handler for signals that work on windows.
81	n/a	# Issue #18396, only for signals without a C-level handler.
82	n/a	if signal.getsignal(sig) is not None:
83	n/a	signal.signal(sig, signal.signal(sig, handler))
84	n/a	checked.add(sig)
85	n/a	# Issue #18396: Ensure the above loop at least tested something
86	n/a	self.assertTrue(checked)
87	n/a
88	n/a	with self.assertRaises(ValueError):
89	n/a	signal.signal(-1, handler)
90	n/a
91	n/a	with self.assertRaises(ValueError):
92	n/a	signal.signal(7, handler)
93	n/a
94	n/a
95	n/a	class WakeupFDTests(unittest.TestCase):
96	n/a
97	n/a	def test_invalid_fd(self):
98	n/a	fd = support.make_bad_fd()
99	n/a	self.assertRaises((ValueError, OSError),
100	n/a	signal.set_wakeup_fd, fd)
101	n/a
102	n/a	def test_invalid_socket(self):
103	n/a	sock = socket.socket()
104	n/a	fd = sock.fileno()
105	n/a	sock.close()
106	n/a	self.assertRaises((ValueError, OSError),
107	n/a	signal.set_wakeup_fd, fd)
108	n/a
109	n/a	def test_set_wakeup_fd_result(self):
110	n/a	r1, w1 = os.pipe()
111	n/a	self.addCleanup(os.close, r1)
112	n/a	self.addCleanup(os.close, w1)
113	n/a	r2, w2 = os.pipe()
114	n/a	self.addCleanup(os.close, r2)
115	n/a	self.addCleanup(os.close, w2)
116	n/a
117	n/a	if hasattr(os, 'set_blocking'):
118	n/a	os.set_blocking(w1, False)
119	n/a	os.set_blocking(w2, False)
120	n/a
121	n/a	signal.set_wakeup_fd(w1)
122	n/a	self.assertEqual(signal.set_wakeup_fd(w2), w1)
123	n/a	self.assertEqual(signal.set_wakeup_fd(-1), w2)
124	n/a	self.assertEqual(signal.set_wakeup_fd(-1), -1)
125	n/a
126	n/a	def test_set_wakeup_fd_socket_result(self):
127	n/a	sock1 = socket.socket()
128	n/a	self.addCleanup(sock1.close)
129	n/a	sock1.setblocking(False)
130	n/a	fd1 = sock1.fileno()
131	n/a
132	n/a	sock2 = socket.socket()
133	n/a	self.addCleanup(sock2.close)
134	n/a	sock2.setblocking(False)
135	n/a	fd2 = sock2.fileno()
136	n/a
137	n/a	signal.set_wakeup_fd(fd1)
138	n/a	self.assertEqual(signal.set_wakeup_fd(fd2), fd1)
139	n/a	self.assertEqual(signal.set_wakeup_fd(-1), fd2)
140	n/a	self.assertEqual(signal.set_wakeup_fd(-1), -1)
141	n/a
142	n/a	# On Windows, files are always blocking and Windows does not provide a
143	n/a	# function to test if a socket is in non-blocking mode.
144	n/a	@unittest.skipIf(sys.platform == "win32", "tests specific to POSIX")
145	n/a	def test_set_wakeup_fd_blocking(self):
146	n/a	rfd, wfd = os.pipe()
147	n/a	self.addCleanup(os.close, rfd)
148	n/a	self.addCleanup(os.close, wfd)
149	n/a
150	n/a	# fd must be non-blocking
151	n/a	os.set_blocking(wfd, True)
152	n/a	with self.assertRaises(ValueError) as cm:
153	n/a	signal.set_wakeup_fd(wfd)
154	n/a	self.assertEqual(str(cm.exception),
155	n/a	"the fd %s must be in non-blocking mode" % wfd)
156	n/a
157	n/a	# non-blocking is ok
158	n/a	os.set_blocking(wfd, False)
159	n/a	signal.set_wakeup_fd(wfd)
160	n/a	signal.set_wakeup_fd(-1)
161	n/a
162	n/a
163	n/a	@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
164	n/a	class WakeupSignalTests(unittest.TestCase):
165	n/a	@unittest.skipIf(_testcapi is None, 'need _testcapi')
166	n/a	def check_wakeup(self, test_body, *signals, ordered=True):
167	n/a	# use a subprocess to have only one thread
168	n/a	code = """if 1:
169	n/a	import _testcapi
170	n/a	import os
171	n/a	import signal
172	n/a	import struct
173	n/a
174	n/a	signals = {!r}
175	n/a
176	n/a	def handler(signum, frame):
177	n/a	pass
178	n/a
179	n/a	def check_signum(signals):
180	n/a	data = os.read(read, len(signals)+1)
181	n/a	raised = struct.unpack('%uB' % len(data), data)
182	n/a	if not {!r}:
183	n/a	raised = set(raised)
184	n/a	signals = set(signals)
185	n/a	if raised != signals:
186	n/a	raise Exception("%r != %r" % (raised, signals))
187	n/a
188	n/a	{}
189	n/a
190	n/a	signal.signal(signal.SIGALRM, handler)
191	n/a	read, write = os.pipe()
192	n/a	os.set_blocking(write, False)
193	n/a	signal.set_wakeup_fd(write)
194	n/a
195	n/a	test()
196	n/a	check_signum(signals)
197	n/a
198	n/a	os.close(read)
199	n/a	os.close(write)
200	n/a	""".format(tuple(map(int, signals)), ordered, test_body)
201	n/a
202	n/a	assert_python_ok('-c', code)
203	n/a
204	n/a	@unittest.skipIf(_testcapi is None, 'need _testcapi')
205	n/a	def test_wakeup_write_error(self):
206	n/a	# Issue #16105: write() errors in the C signal handler should not
207	n/a	# pass silently.
208	n/a	# Use a subprocess to have only one thread.
209	n/a	code = """if 1:
210	n/a	import _testcapi
211	n/a	import errno
212	n/a	import os
213	n/a	import signal
214	n/a	import sys
215	n/a	from test.support import captured_stderr
216	n/a
217	n/a	def handler(signum, frame):
218	n/a	1/0
219	n/a
220	n/a	signal.signal(signal.SIGALRM, handler)
221	n/a	r, w = os.pipe()
222	n/a	os.set_blocking(r, False)
223	n/a
224	n/a	# Set wakeup_fd a read-only file descriptor to trigger the error
225	n/a	signal.set_wakeup_fd(r)
226	n/a	try:
227	n/a	with captured_stderr() as err:
228	n/a	_testcapi.raise_signal(signal.SIGALRM)
229	n/a	except ZeroDivisionError:
230	n/a	# An ignored exception should have been printed out on stderr
231	n/a	err = err.getvalue()
232	n/a	if ('Exception ignored when trying to write to the signal wakeup fd'
233	n/a	not in err):
234	n/a	raise AssertionError(err)
235	n/a	if ('OSError: [Errno %d]' % errno.EBADF) not in err:
236	n/a	raise AssertionError(err)
237	n/a	else:
238	n/a	raise AssertionError("ZeroDivisionError not raised")
239	n/a
240	n/a	os.close(r)
241	n/a	os.close(w)
242	n/a	"""
243	n/a	r, w = os.pipe()
244	n/a	try:
245	n/a	os.write(r, b'x')
246	n/a	except OSError:
247	n/a	pass
248	n/a	else:
249	n/a	self.skipTest("OS doesn't report write() error on the read end of a pipe")
250	n/a	finally:
251	n/a	os.close(r)
252	n/a	os.close(w)
253	n/a
254	n/a	assert_python_ok('-c', code)
255	n/a
256	n/a	def test_wakeup_fd_early(self):
257	n/a	self.check_wakeup("""def test():
258	n/a	import select
259	n/a	import time
260	n/a
261	n/a	TIMEOUT_FULL = 10
262	n/a	TIMEOUT_HALF = 5
263	n/a
264	n/a	class InterruptSelect(Exception):
265	n/a	pass
266	n/a
267	n/a	def handler(signum, frame):
268	n/a	raise InterruptSelect
269	n/a	signal.signal(signal.SIGALRM, handler)
270	n/a
271	n/a	signal.alarm(1)
272	n/a
273	n/a	# We attempt to get a signal during the sleep,
274	n/a	# before select is called
275	n/a	try:
276	n/a	select.select([], [], [], TIMEOUT_FULL)
277	n/a	except InterruptSelect:
278	n/a	pass
279	n/a	else:
280	n/a	raise Exception("select() was not interrupted")
281	n/a
282	n/a	before_time = time.monotonic()
283	n/a	select.select([read], [], [], TIMEOUT_FULL)
284	n/a	after_time = time.monotonic()
285	n/a	dt = after_time - before_time
286	n/a	if dt >= TIMEOUT_HALF:
287	n/a	raise Exception("%s >= %s" % (dt, TIMEOUT_HALF))
288	n/a	""", signal.SIGALRM)
289	n/a
290	n/a	def test_wakeup_fd_during(self):
291	n/a	self.check_wakeup("""def test():
292	n/a	import select
293	n/a	import time
294	n/a
295	n/a	TIMEOUT_FULL = 10
296	n/a	TIMEOUT_HALF = 5
297	n/a
298	n/a	class InterruptSelect(Exception):
299	n/a	pass
300	n/a
301	n/a	def handler(signum, frame):
302	n/a	raise InterruptSelect
303	n/a	signal.signal(signal.SIGALRM, handler)
304	n/a
305	n/a	signal.alarm(1)
306	n/a	before_time = time.monotonic()
307	n/a	# We attempt to get a signal during the select call
308	n/a	try:
309	n/a	select.select([read], [], [], TIMEOUT_FULL)
310	n/a	except InterruptSelect:
311	n/a	pass
312	n/a	else:
313	n/a	raise Exception("select() was not interrupted")
314	n/a	after_time = time.monotonic()
315	n/a	dt = after_time - before_time
316	n/a	if dt >= TIMEOUT_HALF:
317	n/a	raise Exception("%s >= %s" % (dt, TIMEOUT_HALF))
318	n/a	""", signal.SIGALRM)
319	n/a
320	n/a	def test_signum(self):
321	n/a	self.check_wakeup("""def test():
322	n/a	import _testcapi
323	n/a	signal.signal(signal.SIGUSR1, handler)
324	n/a	_testcapi.raise_signal(signal.SIGUSR1)
325	n/a	_testcapi.raise_signal(signal.SIGALRM)
326	n/a	""", signal.SIGUSR1, signal.SIGALRM)
327	n/a
328	n/a	@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
329	n/a	'need signal.pthread_sigmask()')
330	n/a	def test_pending(self):
331	n/a	self.check_wakeup("""def test():
332	n/a	signum1 = signal.SIGUSR1
333	n/a	signum2 = signal.SIGUSR2
334	n/a
335	n/a	signal.signal(signum1, handler)
336	n/a	signal.signal(signum2, handler)
337	n/a
338	n/a	signal.pthread_sigmask(signal.SIG_BLOCK, (signum1, signum2))
339	n/a	_testcapi.raise_signal(signum1)
340	n/a	_testcapi.raise_signal(signum2)
341	n/a	# Unblocking the 2 signals calls the C signal handler twice
342	n/a	signal.pthread_sigmask(signal.SIG_UNBLOCK, (signum1, signum2))
343	n/a	""", signal.SIGUSR1, signal.SIGUSR2, ordered=False)
344	n/a
345	n/a
346	n/a	@unittest.skipUnless(hasattr(socket, 'socketpair'), 'need socket.socketpair')
347	n/a	class WakeupSocketSignalTests(unittest.TestCase):
348	n/a
349	n/a	@unittest.skipIf(_testcapi is None, 'need _testcapi')
350	n/a	def test_socket(self):
351	n/a	# use a subprocess to have only one thread
352	n/a	code = """if 1:
353	n/a	import signal
354	n/a	import socket
355	n/a	import struct
356	n/a	import _testcapi
357	n/a
358	n/a	signum = signal.SIGINT
359	n/a	signals = (signum,)
360	n/a
361	n/a	def handler(signum, frame):
362	n/a	pass
363	n/a
364	n/a	signal.signal(signum, handler)
365	n/a
366	n/a	read, write = socket.socketpair()
367	n/a	read.setblocking(False)
368	n/a	write.setblocking(False)
369	n/a	signal.set_wakeup_fd(write.fileno())
370	n/a
371	n/a	_testcapi.raise_signal(signum)
372	n/a
373	n/a	data = read.recv(1)
374	n/a	if not data:
375	n/a	raise Exception("no signum written")
376	n/a	raised = struct.unpack('B', data)
377	n/a	if raised != signals:
378	n/a	raise Exception("%r != %r" % (raised, signals))
379	n/a
380	n/a	read.close()
381	n/a	write.close()
382	n/a	"""
383	n/a
384	n/a	assert_python_ok('-c', code)
385	n/a
386	n/a	@unittest.skipIf(_testcapi is None, 'need _testcapi')
387	n/a	def test_send_error(self):
388	n/a	# Use a subprocess to have only one thread.
389	n/a	if os.name == 'nt':
390	n/a	action = 'send'
391	n/a	else:
392	n/a	action = 'write'
393	n/a	code = """if 1:
394	n/a	import errno
395	n/a	import signal
396	n/a	import socket
397	n/a	import sys
398	n/a	import time
399	n/a	import _testcapi
400	n/a	from test.support import captured_stderr
401	n/a
402	n/a	signum = signal.SIGINT
403	n/a
404	n/a	def handler(signum, frame):
405	n/a	pass
406	n/a
407	n/a	signal.signal(signum, handler)
408	n/a
409	n/a	read, write = socket.socketpair()
410	n/a	read.setblocking(False)
411	n/a	write.setblocking(False)
412	n/a
413	n/a	signal.set_wakeup_fd(write.fileno())
414	n/a
415	n/a	# Close sockets: send() will fail
416	n/a	read.close()
417	n/a	write.close()
418	n/a
419	n/a	with captured_stderr() as err:
420	n/a	_testcapi.raise_signal(signum)
421	n/a
422	n/a	err = err.getvalue()
423	n/a	if ('Exception ignored when trying to {action} to the signal wakeup fd'
424	n/a	not in err):
425	n/a	raise AssertionError(err)
426	n/a	""".format(action=action)
427	n/a	assert_python_ok('-c', code)
428	n/a
429	n/a
430	n/a	@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
431	n/a	class SiginterruptTest(unittest.TestCase):
432	n/a
433	n/a	def readpipe_interrupted(self, interrupt):
434	n/a	"""Perform a read during which a signal will arrive. Return True if the
435	n/a	read is interrupted by the signal and raises an exception. Return False
436	n/a	if it returns normally.
437	n/a	"""
438	n/a	# use a subprocess to have only one thread, to have a timeout on the
439	n/a	# blocking read and to not touch signal handling in this process
440	n/a	code = """if 1:
441	n/a	import errno
442	n/a	import os
443	n/a	import signal
444	n/a	import sys
445	n/a
446	n/a	interrupt = %r
447	n/a	r, w = os.pipe()
448	n/a
449	n/a	def handler(signum, frame):
450	n/a	1 / 0
451	n/a
452	n/a	signal.signal(signal.SIGALRM, handler)
453	n/a	if interrupt is not None:
454	n/a	signal.siginterrupt(signal.SIGALRM, interrupt)
455	n/a
456	n/a	print("ready")
457	n/a	sys.stdout.flush()
458	n/a
459	n/a	# run the test twice
460	n/a	try:
461	n/a	for loop in range(2):
462	n/a	# send a SIGALRM in a second (during the read)
463	n/a	signal.alarm(1)
464	n/a	try:
465	n/a	# blocking call: read from a pipe without data
466	n/a	os.read(r, 1)
467	n/a	except ZeroDivisionError:
468	n/a	pass
469	n/a	else:
470	n/a	sys.exit(2)
471	n/a	sys.exit(3)
472	n/a	finally:
473	n/a	os.close(r)
474	n/a	os.close(w)
475	n/a	""" % (interrupt,)
476	n/a	with spawn_python('-c', code) as process:
477	n/a	try:
478	n/a	# wait until the child process is loaded and has started
479	n/a	first_line = process.stdout.readline()
480	n/a
481	n/a	stdout, stderr = process.communicate(timeout=5.0)
482	n/a	except subprocess.TimeoutExpired:
483	n/a	process.kill()
484	n/a	return False
485	n/a	else:
486	n/a	stdout = first_line + stdout
487	n/a	exitcode = process.wait()
488	n/a	if exitcode not in (2, 3):
489	n/a	raise Exception("Child error (exit code %s): %r"
490	n/a	% (exitcode, stdout))
491	n/a	return (exitcode == 3)
492	n/a
493	n/a	def test_without_siginterrupt(self):
494	n/a	# If a signal handler is installed and siginterrupt is not called
495	n/a	# at all, when that signal arrives, it interrupts a syscall that's in
496	n/a	# progress.
497	n/a	interrupted = self.readpipe_interrupted(None)
498	n/a	self.assertTrue(interrupted)
499	n/a
500	n/a	def test_siginterrupt_on(self):
501	n/a	# If a signal handler is installed and siginterrupt is called with
502	n/a	# a true value for the second argument, when that signal arrives, it
503	n/a	# interrupts a syscall that's in progress.
504	n/a	interrupted = self.readpipe_interrupted(True)
505	n/a	self.assertTrue(interrupted)
506	n/a
507	n/a	def test_siginterrupt_off(self):
508	n/a	# If a signal handler is installed and siginterrupt is called with
509	n/a	# a false value for the second argument, when that signal arrives, it
510	n/a	# does not interrupt a syscall that's in progress.
511	n/a	interrupted = self.readpipe_interrupted(False)
512	n/a	self.assertFalse(interrupted)
513	n/a
514	n/a
515	n/a	@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
516	n/a	class ItimerTest(unittest.TestCase):
517	n/a	def setUp(self):
518	n/a	self.hndl_called = False
519	n/a	self.hndl_count = 0
520	n/a	self.itimer = None
521	n/a	self.old_alarm = signal.signal(signal.SIGALRM, self.sig_alrm)
522	n/a
523	n/a	def tearDown(self):
524	n/a	signal.signal(signal.SIGALRM, self.old_alarm)
525	n/a	if self.itimer is not None: # test_itimer_exc doesn't change this attr
526	n/a	# just ensure that itimer is stopped
527	n/a	signal.setitimer(self.itimer, 0)
528	n/a
529	n/a	def sig_alrm(self, *args):
530	n/a	self.hndl_called = True
531	n/a
532	n/a	def sig_vtalrm(self, *args):
533	n/a	self.hndl_called = True
534	n/a
535	n/a	if self.hndl_count > 3:
536	n/a	# it shouldn't be here, because it should have been disabled.
537	n/a	raise signal.ItimerError("setitimer didn't disable ITIMER_VIRTUAL "
538	n/a	"timer.")
539	n/a	elif self.hndl_count == 3:
540	n/a	# disable ITIMER_VIRTUAL, this function shouldn't be called anymore
541	n/a	signal.setitimer(signal.ITIMER_VIRTUAL, 0)
542	n/a
543	n/a	self.hndl_count += 1
544	n/a
545	n/a	def sig_prof(self, *args):
546	n/a	self.hndl_called = True
547	n/a	signal.setitimer(signal.ITIMER_PROF, 0)
548	n/a
549	n/a	def test_itimer_exc(self):
550	n/a	# XXX I'm assuming -1 is an invalid itimer, but maybe some platform
551	n/a	# defines it ?
552	n/a	self.assertRaises(signal.ItimerError, signal.setitimer, -1, 0)
553	n/a	# Negative times are treated as zero on some platforms.
554	n/a	if 0:
555	n/a	self.assertRaises(signal.ItimerError,
556	n/a	signal.setitimer, signal.ITIMER_REAL, -1)
557	n/a
558	n/a	def test_itimer_real(self):
559	n/a	self.itimer = signal.ITIMER_REAL
560	n/a	signal.setitimer(self.itimer, 1.0)
561	n/a	signal.pause()
562	n/a	self.assertEqual(self.hndl_called, True)
563	n/a
564	n/a	# Issue 3864, unknown if this affects earlier versions of freebsd also
565	n/a	@unittest.skipIf(sys.platform in ('freebsd6', 'netbsd5'),
566	n/a	'itimer not reliable (does not mix well with threading) on some BSDs.')
567	n/a	def test_itimer_virtual(self):
568	n/a	self.itimer = signal.ITIMER_VIRTUAL
569	n/a	signal.signal(signal.SIGVTALRM, self.sig_vtalrm)
570	n/a	signal.setitimer(self.itimer, 0.3, 0.2)
571	n/a
572	n/a	start_time = time.monotonic()
573	n/a	while time.monotonic() - start_time < 60.0:
574	n/a	# use up some virtual time by doing real work
575	n/a	_ = pow(12345, 67890, 10000019)
576	n/a	if signal.getitimer(self.itimer) == (0.0, 0.0):
577	n/a	break # sig_vtalrm handler stopped this itimer
578	n/a	else: # Issue 8424
579	n/a	self.skipTest("timeout: likely cause: machine too slow or load too "
580	n/a	"high")
581	n/a
582	n/a	# virtual itimer should be (0.0, 0.0) now
583	n/a	self.assertEqual(signal.getitimer(self.itimer), (0.0, 0.0))
584	n/a	# and the handler should have been called
585	n/a	self.assertEqual(self.hndl_called, True)
586	n/a
587	n/a	# Issue 3864, unknown if this affects earlier versions of freebsd also
588	n/a	@unittest.skipIf(sys.platform=='freebsd6',
589	n/a	'itimer not reliable (does not mix well with threading) on freebsd6')
590	n/a	def test_itimer_prof(self):
591	n/a	self.itimer = signal.ITIMER_PROF
592	n/a	signal.signal(signal.SIGPROF, self.sig_prof)
593	n/a	signal.setitimer(self.itimer, 0.2, 0.2)
594	n/a
595	n/a	start_time = time.monotonic()
596	n/a	while time.monotonic() - start_time < 60.0:
597	n/a	# do some work
598	n/a	_ = pow(12345, 67890, 10000019)
599	n/a	if signal.getitimer(self.itimer) == (0.0, 0.0):
600	n/a	break # sig_prof handler stopped this itimer
601	n/a	else: # Issue 8424
602	n/a	self.skipTest("timeout: likely cause: machine too slow or load too "
603	n/a	"high")
604	n/a
605	n/a	# profiling itimer should be (0.0, 0.0) now
606	n/a	self.assertEqual(signal.getitimer(self.itimer), (0.0, 0.0))
607	n/a	# and the handler should have been called
608	n/a	self.assertEqual(self.hndl_called, True)
609	n/a
610	n/a
611	n/a	class PendingSignalsTests(unittest.TestCase):
612	n/a	"""
613	n/a	Test pthread_sigmask(), pthread_kill(), sigpending() and sigwait()
614	n/a	functions.
615	n/a	"""
616	n/a	@unittest.skipUnless(hasattr(signal, 'sigpending'),
617	n/a	'need signal.sigpending()')
618	n/a	def test_sigpending_empty(self):
619	n/a	self.assertEqual(signal.sigpending(), set())
620	n/a
621	n/a	@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
622	n/a	'need signal.pthread_sigmask()')
623	n/a	@unittest.skipUnless(hasattr(signal, 'sigpending'),
624	n/a	'need signal.sigpending()')
625	n/a	def test_sigpending(self):
626	n/a	code = """if 1:
627	n/a	import os
628	n/a	import signal
629	n/a
630	n/a	def handler(signum, frame):
631	n/a	1/0
632	n/a
633	n/a	signum = signal.SIGUSR1
634	n/a	signal.signal(signum, handler)
635	n/a
636	n/a	signal.pthread_sigmask(signal.SIG_BLOCK, [signum])
637	n/a	os.kill(os.getpid(), signum)
638	n/a	pending = signal.sigpending()
639	n/a	for sig in pending:
640	n/a	assert isinstance(sig, signal.Signals), repr(pending)
641	n/a	if pending != {signum}:
642	n/a	raise Exception('%s != {%s}' % (pending, signum))
643	n/a	try:
644	n/a	signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum])
645	n/a	except ZeroDivisionError:
646	n/a	pass
647	n/a	else:
648	n/a	raise Exception("ZeroDivisionError not raised")
649	n/a	"""
650	n/a	assert_python_ok('-c', code)
651	n/a
652	n/a	@unittest.skipUnless(hasattr(signal, 'pthread_kill'),
653	n/a	'need signal.pthread_kill()')
654	n/a	def test_pthread_kill(self):
655	n/a	code = """if 1:
656	n/a	import signal
657	n/a	import threading
658	n/a	import sys
659	n/a
660	n/a	signum = signal.SIGUSR1
661	n/a
662	n/a	def handler(signum, frame):
663	n/a	1/0
664	n/a
665	n/a	signal.signal(signum, handler)
666	n/a
667	n/a	if sys.platform == 'freebsd6':
668	n/a	# Issue #12392 and #12469: send a signal to the main thread
669	n/a	# doesn't work before the creation of the first thread on
670	n/a	# FreeBSD 6
671	n/a	def noop():
672	n/a	pass
673	n/a	thread = threading.Thread(target=noop)
674	n/a	thread.start()
675	n/a	thread.join()
676	n/a
677	n/a	tid = threading.get_ident()
678	n/a	try:
679	n/a	signal.pthread_kill(tid, signum)
680	n/a	except ZeroDivisionError:
681	n/a	pass
682	n/a	else:
683	n/a	raise Exception("ZeroDivisionError not raised")
684	n/a	"""
685	n/a	assert_python_ok('-c', code)
686	n/a
687	n/a	@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
688	n/a	'need signal.pthread_sigmask()')
689	n/a	def wait_helper(self, blocked, test):
690	n/a	"""
691	n/a	test: body of the "def test(signum):" function.
692	n/a	blocked: number of the blocked signal
693	n/a	"""
694	n/a	code = '''if 1:
695	n/a	import signal
696	n/a	import sys
697	n/a	from signal import Signals
698	n/a
699	n/a	def handler(signum, frame):
700	n/a	1/0
701	n/a
702	n/a	%s
703	n/a
704	n/a	blocked = %s
705	n/a	signum = signal.SIGALRM
706	n/a
707	n/a	# child: block and wait the signal
708	n/a	try:
709	n/a	signal.signal(signum, handler)
710	n/a	signal.pthread_sigmask(signal.SIG_BLOCK, [blocked])
711	n/a
712	n/a	# Do the tests
713	n/a	test(signum)
714	n/a
715	n/a	# The handler must not be called on unblock
716	n/a	try:
717	n/a	signal.pthread_sigmask(signal.SIG_UNBLOCK, [blocked])
718	n/a	except ZeroDivisionError:
719	n/a	print("the signal handler has been called",
720	n/a	file=sys.stderr)
721	n/a	sys.exit(1)
722	n/a	except BaseException as err:
723	n/a	print("error: {}".format(err), file=sys.stderr)
724	n/a	sys.stderr.flush()
725	n/a	sys.exit(1)
726	n/a	''' % (test.strip(), blocked)
727	n/a
728	n/a	# sigwait must be called with the signal blocked: since the current
729	n/a	# process might have several threads running, use a subprocess to have
730	n/a	# a single thread.
731	n/a	assert_python_ok('-c', code)
732	n/a
733	n/a	@unittest.skipUnless(hasattr(signal, 'sigwait'),
734	n/a	'need signal.sigwait()')
735	n/a	def test_sigwait(self):
736	n/a	self.wait_helper(signal.SIGALRM, '''
737	n/a	def test(signum):
738	n/a	signal.alarm(1)
739	n/a	received = signal.sigwait([signum])
740	n/a	assert isinstance(received, signal.Signals), received
741	n/a	if received != signum:
742	n/a	raise Exception('received %s, not %s' % (received, signum))
743	n/a	''')
744	n/a
745	n/a	@unittest.skipUnless(hasattr(signal, 'sigwaitinfo'),
746	n/a	'need signal.sigwaitinfo()')
747	n/a	def test_sigwaitinfo(self):
748	n/a	self.wait_helper(signal.SIGALRM, '''
749	n/a	def test(signum):
750	n/a	signal.alarm(1)
751	n/a	info = signal.sigwaitinfo([signum])
752	n/a	if info.si_signo != signum:
753	n/a	raise Exception("info.si_signo != %s" % signum)
754	n/a	''')
755	n/a
756	n/a	@unittest.skipUnless(hasattr(signal, 'sigtimedwait'),
757	n/a	'need signal.sigtimedwait()')
758	n/a	def test_sigtimedwait(self):
759	n/a	self.wait_helper(signal.SIGALRM, '''
760	n/a	def test(signum):
761	n/a	signal.alarm(1)
762	n/a	info = signal.sigtimedwait([signum], 10.1000)
763	n/a	if info.si_signo != signum:
764	n/a	raise Exception('info.si_signo != %s' % signum)
765	n/a	''')
766	n/a
767	n/a	@unittest.skipUnless(hasattr(signal, 'sigtimedwait'),
768	n/a	'need signal.sigtimedwait()')
769	n/a	def test_sigtimedwait_poll(self):
770	n/a	# check that polling with sigtimedwait works
771	n/a	self.wait_helper(signal.SIGALRM, '''
772	n/a	def test(signum):
773	n/a	import os
774	n/a	os.kill(os.getpid(), signum)
775	n/a	info = signal.sigtimedwait([signum], 0)
776	n/a	if info.si_signo != signum:
777	n/a	raise Exception('info.si_signo != %s' % signum)
778	n/a	''')
779	n/a
780	n/a	@unittest.skipUnless(hasattr(signal, 'sigtimedwait'),
781	n/a	'need signal.sigtimedwait()')
782	n/a	def test_sigtimedwait_timeout(self):
783	n/a	self.wait_helper(signal.SIGALRM, '''
784	n/a	def test(signum):
785	n/a	received = signal.sigtimedwait([signum], 1.0)
786	n/a	if received is not None:
787	n/a	raise Exception("received=%r" % (received,))
788	n/a	''')
789	n/a
790	n/a	@unittest.skipUnless(hasattr(signal, 'sigtimedwait'),
791	n/a	'need signal.sigtimedwait()')
792	n/a	def test_sigtimedwait_negative_timeout(self):
793	n/a	signum = signal.SIGALRM
794	n/a	self.assertRaises(ValueError, signal.sigtimedwait, [signum], -1.0)
795	n/a
796	n/a	@unittest.skipUnless(hasattr(signal, 'sigwait'),
797	n/a	'need signal.sigwait()')
798	n/a	@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
799	n/a	'need signal.pthread_sigmask()')
800	n/a	@unittest.skipIf(threading is None, "test needs threading module")
801	n/a	def test_sigwait_thread(self):
802	n/a	# Check that calling sigwait() from a thread doesn't suspend the whole
803	n/a	# process. A new interpreter is spawned to avoid problems when mixing
804	n/a	# threads and fork(): only async-safe functions are allowed between
805	n/a	# fork() and exec().
806	n/a	assert_python_ok("-c", """if True:
807	n/a	import os, threading, sys, time, signal
808	n/a
809	n/a	# the default handler terminates the process
810	n/a	signum = signal.SIGUSR1
811	n/a
812	n/a	def kill_later():
813	n/a	# wait until the main thread is waiting in sigwait()
814	n/a	time.sleep(1)
815	n/a	os.kill(os.getpid(), signum)
816	n/a
817	n/a	# the signal must be blocked by all the threads
818	n/a	signal.pthread_sigmask(signal.SIG_BLOCK, [signum])
819	n/a	killer = threading.Thread(target=kill_later)
820	n/a	killer.start()
821	n/a	received = signal.sigwait([signum])
822	n/a	if received != signum:
823	n/a	print("sigwait() received %s, not %s" % (received, signum),
824	n/a	file=sys.stderr)
825	n/a	sys.exit(1)
826	n/a	killer.join()
827	n/a	# unblock the signal, which should have been cleared by sigwait()
828	n/a	signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum])
829	n/a	""")
830	n/a
831	n/a	@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
832	n/a	'need signal.pthread_sigmask()')
833	n/a	def test_pthread_sigmask_arguments(self):
834	n/a	self.assertRaises(TypeError, signal.pthread_sigmask)
835	n/a	self.assertRaises(TypeError, signal.pthread_sigmask, 1)
836	n/a	self.assertRaises(TypeError, signal.pthread_sigmask, 1, 2, 3)
837	n/a	self.assertRaises(OSError, signal.pthread_sigmask, 1700, [])
838	n/a
839	n/a	@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
840	n/a	'need signal.pthread_sigmask()')
841	n/a	def test_pthread_sigmask(self):
842	n/a	code = """if 1:
843	n/a	import signal
844	n/a	import os; import threading
845	n/a
846	n/a	def handler(signum, frame):
847	n/a	1/0
848	n/a
849	n/a	def kill(signum):
850	n/a	os.kill(os.getpid(), signum)
851	n/a
852	n/a	def check_mask(mask):
853	n/a	for sig in mask:
854	n/a	assert isinstance(sig, signal.Signals), repr(sig)
855	n/a
856	n/a	def read_sigmask():
857	n/a	sigmask = signal.pthread_sigmask(signal.SIG_BLOCK, [])
858	n/a	check_mask(sigmask)
859	n/a	return sigmask
860	n/a
861	n/a	signum = signal.SIGUSR1
862	n/a
863	n/a	# Install our signal handler
864	n/a	old_handler = signal.signal(signum, handler)
865	n/a
866	n/a	# Unblock SIGUSR1 (and copy the old mask) to test our signal handler
867	n/a	old_mask = signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum])
868	n/a	check_mask(old_mask)
869	n/a	try:
870	n/a	kill(signum)
871	n/a	except ZeroDivisionError:
872	n/a	pass
873	n/a	else:
874	n/a	raise Exception("ZeroDivisionError not raised")
875	n/a
876	n/a	# Block and then raise SIGUSR1. The signal is blocked: the signal
877	n/a	# handler is not called, and the signal is now pending
878	n/a	mask = signal.pthread_sigmask(signal.SIG_BLOCK, [signum])
879	n/a	check_mask(mask)
880	n/a	kill(signum)
881	n/a
882	n/a	# Check the new mask
883	n/a	blocked = read_sigmask()
884	n/a	check_mask(blocked)
885	n/a	if signum not in blocked:
886	n/a	raise Exception("%s not in %s" % (signum, blocked))
887	n/a	if old_mask ^ blocked != {signum}:
888	n/a	raise Exception("%s ^ %s != {%s}" % (old_mask, blocked, signum))
889	n/a
890	n/a	# Unblock SIGUSR1
891	n/a	try:
892	n/a	# unblock the pending signal calls immediately the signal handler
893	n/a	signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum])
894	n/a	except ZeroDivisionError:
895	n/a	pass
896	n/a	else:
897	n/a	raise Exception("ZeroDivisionError not raised")
898	n/a	try:
899	n/a	kill(signum)
900	n/a	except ZeroDivisionError:
901	n/a	pass
902	n/a	else:
903	n/a	raise Exception("ZeroDivisionError not raised")
904	n/a
905	n/a	# Check the new mask
906	n/a	unblocked = read_sigmask()
907	n/a	if signum in unblocked:
908	n/a	raise Exception("%s in %s" % (signum, unblocked))
909	n/a	if blocked ^ unblocked != {signum}:
910	n/a	raise Exception("%s ^ %s != {%s}" % (blocked, unblocked, signum))
911	n/a	if old_mask != unblocked:
912	n/a	raise Exception("%s != %s" % (old_mask, unblocked))
913	n/a	"""
914	n/a	assert_python_ok('-c', code)
915	n/a
916	n/a	@unittest.skipIf(sys.platform == 'freebsd6',
917	n/a	"issue #12392: send a signal to the main thread doesn't work "
918	n/a	"before the creation of the first thread on FreeBSD 6")
919	n/a	@unittest.skipUnless(hasattr(signal, 'pthread_kill'),
920	n/a	'need signal.pthread_kill()')
921	n/a	def test_pthread_kill_main_thread(self):
922	n/a	# Test that a signal can be sent to the main thread with pthread_kill()
923	n/a	# before any other thread has been created (see issue #12392).
924	n/a	code = """if True:
925	n/a	import threading
926	n/a	import signal
927	n/a	import sys
928	n/a
929	n/a	def handler(signum, frame):
930	n/a	sys.exit(3)
931	n/a
932	n/a	signal.signal(signal.SIGUSR1, handler)
933	n/a	signal.pthread_kill(threading.get_ident(), signal.SIGUSR1)
934	n/a	sys.exit(2)
935	n/a	"""
936	n/a
937	n/a	with spawn_python('-c', code) as process:
938	n/a	stdout, stderr = process.communicate()
939	n/a	exitcode = process.wait()
940	n/a	if exitcode != 3:
941	n/a	raise Exception("Child error (exit code %s): %s" %
942	n/a	(exitcode, stdout))
943	n/a
944	n/a
945	n/a	def tearDownModule():
946	n/a	support.reap_children()
947	n/a
948	n/a	if __name__ == "__main__":
949	n/a	unittest.main()