Skip to content

C
cpython
Commit 07e536f4 authored by Georg Brandl's avatar Georg Brandl
Browse files
Options

Patch #1008: port test_signal to unittest.

parent c56de5f4
Hide whitespace changes
Inline Side-by-side
Showing with 171 additions and 163 deletions
Lib/test/output/test_signal deleted 100644 → 0
View file @ c56de5f4
test_signal
starting pause() loop...
Lib/test/test_signal.py
View file @ 07e536f4
# Test the signal module import unittest
from test.test_support import verbose, TestSkipped, TestFailed, vereq from test import test_support
import signal import signal
import os, sys, time import os, sys, time
if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos':
raise TestSkipped, "Can't test signal on %s" % sys.platform
MAX_DURATION = 20 # Entire test should last at most 20 sec.
if verbose:
x = '-x'
else:
x = '+x'
pid = os.getpid()
if verbose:
print "test runner's pid is", pid
# Shell script that will send us asynchronous signals
script = """
(
set %(x)s
sleep 2
kill -HUP %(pid)d
sleep 2
kill -USR1 %(pid)d
sleep 2
kill -USR2 %(pid)d
) &
""" % vars()
a_called = b_called = False
def handlerA(*args):
global a_called
a_called = True
if verbose:
print "handlerA invoked", args
class HandlerBCalled(Exception): class HandlerBCalled(Exception):
pass pass
def handlerB(*args): class InterProcessSignalTests(unittest.TestCase):
global b_called MAX_DURATION = 20 # Entire test should last at most 20 sec.
b_called = True
if verbose: # Set up a child to send signals to us (the parent) after waiting
print "handlerB invoked", args # long enough to receive the alarm. It seems we miss the alarm
raise HandlerBCalled, args # for some reason. This will hopefully stop the hangs on
# Tru64/Alpha. Alas, it doesn't. Tru64 appears to miss all the
# Set up a child to send signals to us (the parent) after waiting long # signals at times, or seemingly random subsets of them, and
# enough to receive the alarm. It seems we miss the alarm for some # nothing done in force_test_exit so far has actually helped.
# reason. This will hopefully stop the hangs on Tru64/Alpha. def spawn_force_test_exit_process(self, parent_pid):
# Alas, it doesn't. Tru64 appears to miss all the signals at times, or # Sigh, both imports seem necessary to avoid errors.
# seemingly random subsets of them, and nothing done in force_test_exit import os
# so far has actually helped. fork_pid = os.fork()
def force_test_exit(): if fork_pid:
# Sigh, both imports seem necessary to avoid errors. # In parent.
import os return fork_pid
fork_pid = os.fork()
if fork_pid: # In child.
# In parent. import os, time
return fork_pid try:
# Wait 5 seconds longer than the expected alarm to give enough
# In child. # time for the normal sequence of events to occur. This is
import os, time # just a stop-gap to try to prevent the test from hanging.
try: time.sleep(self.MAX_DURATION + 5)
# Wait 5 seconds longer than the expected alarm to give enough print >> sys.__stdout__, " child should not have to kill parent"
# time for the normal sequence of events to occur. This is for signame in "SIGHUP", "SIGUSR1", "SIGUSR2", "SIGALRM":
# just a stop-gap to try to prevent the test from hanging. os.kill(parent_pid, getattr(signal, signame))
time.sleep(MAX_DURATION + 5) print >> sys.__stdout__, " child sent", signame, "to", \
print >> sys.__stdout__, ' child should not have to kill parent' parent_pid
for signame in "SIGHUP", "SIGUSR1", "SIGUSR2", "SIGALRM": time.sleep(1)
os.kill(pid, getattr(signal, signame)) finally:
print >> sys.__stdout__, " child sent", signame, "to", pid os._exit(0)
time.sleep(1)
finally: def handlerA(self, *args):
os._exit(0) self.a_called = True
if test_support.verbose:
# Install handlers. print "handlerA invoked", args
hup = signal.signal(signal.SIGHUP, handlerA)
usr1 = signal.signal(signal.SIGUSR1, handlerB) def handlerB(self, *args):
usr2 = signal.signal(signal.SIGUSR2, signal.SIG_IGN) self.b_called = True
alrm = signal.signal(signal.SIGALRM, signal.default_int_handler) if test_support.verbose:
print "handlerB invoked", args
try: raise HandlerBCalled(*args)
signal.alarm(MAX_DURATION) def test_main(self):
vereq(signal.getsignal(signal.SIGHUP), handlerA) self.assertEquals(signal.getsignal(signal.SIGHUP), self.handlerA)
vereq(signal.getsignal(signal.SIGUSR1), handlerB) self.assertEquals(signal.getsignal(signal.SIGUSR1), self.handlerB)
vereq(signal.getsignal(signal.SIGUSR2), signal.SIG_IGN) self.assertEquals(signal.getsignal(signal.SIGUSR2), signal.SIG_IGN)
vereq(signal.getsignal(signal.SIGALRM), signal.default_int_handler) self.assertEquals(signal.getsignal(signal.SIGALRM),
signal.default_int_handler)
# Try to ensure this test exits even if there is some problem with alarm.
# Tru64/Alpha often hangs and is ultimately killed by the buildbot. # Launch an external script to send us signals.
fork_pid = force_test_exit() # We expect the external script to:
# send HUP, which invokes handlerA to set a_called
try: # send USR1, which invokes handlerB to set b_called and raise
signal.getsignal(4242) # HandlerBCalled
raise TestFailed('expected ValueError for invalid signal # to ' # send USR2, which is ignored
'getsignal()') #
except ValueError: # Then we expect the alarm to go off, and its handler raises
pass # KeyboardInterrupt, finally getting us out of the loop.
try: if test_support.verbose:
signal.signal(4242, handlerB) verboseflag = '-x'
raise TestFailed('expected ValueError for invalid signal # to ' else:
'signal()') verboseflag = '+x'
except ValueError:
pass pid = self.pid
if test_support.verbose:
try: print "test runner's pid is", pid
signal.signal(signal.SIGUSR1, None)
raise TestFailed('expected TypeError for non-callable') # Shell script that will send us asynchronous signals
except TypeError: script = """
pass (
set %(verboseflag)s
# Launch an external script to send us signals. sleep 2
# We expect the external script to: kill -HUP %(pid)d
# send HUP, which invokes handlerA to set a_called sleep 2
# send USR1, which invokes handlerB to set b_called and raise kill -USR1 %(pid)d
# HandlerBCalled sleep 2
# send USR2, which is ignored kill -USR2 %(pid)d
# ) &
# Then we expect the alarm to go off, and its handler raises """ % vars()
# KeyboardInterrupt, finally getting us out of the loop.
os.system(script) signal.alarm(self.MAX_DURATION)
try:
print "starting pause() loop..." handler_b_exception_raised = False
while 1:
try: os.system(script)
if verbose: try:
print "call pause()..." if test_support.verbose:
signal.pause() print "starting pause() loop..."
if verbose: while 1:
print "pause() returned" try:
except HandlerBCalled: if test_support.verbose:
if verbose: print "call pause()..."
print "HandlerBCalled exception caught" signal.pause()
if test_support.verbose:
except KeyboardInterrupt: print "pause() returned"
if verbose: except HandlerBCalled:
print "KeyboardInterrupt (the alarm() went off)" handler_b_exception_raised = True
if test_support.verbose:
if not a_called: print "HandlerBCalled exception caught"
print 'HandlerA not called'
except KeyboardInterrupt:
if not b_called: if test_support.verbose:
print 'HandlerB not called' print "KeyboardInterrupt (the alarm() went off)"
finally: self.assert_(self.a_called)
# Forcibly kill the child we created to ping us if there was a test error. self.assert_(self.b_called)
try: self.assert_(handler_b_exception_raised)
# Make sure we don't kill ourself if there was a fork error.
if fork_pid > 0: def setUp(self):
os.kill(fork_pid, signal.SIGKILL) # Install handlers.
except: self.hup = signal.signal(signal.SIGHUP, self.handlerA)
# If the child killed us, it has probably exited. Killing a self.usr1 = signal.signal(signal.SIGUSR1, self.handlerB)
# non-existent process will raise an error which we don't care about. self.usr2 = signal.signal(signal.SIGUSR2, signal.SIG_IGN)
pass self.alrm = signal.signal(signal.SIGALRM,
signal.default_int_handler)
# Restore handlers. self.a_called = False
signal.alarm(0) # cancel alarm in case we died early self.b_called = False
signal.signal(signal.SIGHUP, hup) self.pid = os.getpid()
signal.signal(signal.SIGUSR1, usr1) self.fork_pid = self.spawn_force_test_exit_process(self.pid)
signal.signal(signal.SIGUSR2, usr2)
signal.signal(signal.SIGALRM, alrm) def tearDown(self):
# Forcibly kill the child we created to ping us if there was a
# test error.
try:
# Make sure we don't kill ourself if there was a fork
# error.
if self.fork_pid > 0:
os.kill(self.fork_pid, signal.SIGKILL)
except:
# If the child killed us, it has probably exited. Killing
# a non-existent process will raise an error which we
# don't care about.
pass
# Restore handlers.
signal.alarm(0) # cancel alarm in case we died early
signal.signal(signal.SIGHUP, self.hup)
signal.signal(signal.SIGUSR1, self.usr1)
signal.signal(signal.SIGUSR2, self.usr2)
signal.signal(signal.SIGALRM, self.alrm)
class BasicSignalTests(unittest.TestCase):
def test_out_of_range_signal_number_raises_error(self):
self.assertRaises(ValueError, signal.getsignal, 4242)
def trivial_signal_handler(*args):
pass
self.assertRaises(ValueError, signal.signal, 4242,
trivial_signal_handler)
def test_setting_signal_handler_to_none_raises_error(self):
self.assertRaises(TypeError, signal.signal,
signal.SIGUSR1, None)
def test_main():
if sys.platform[:3] in ('win', 'os2') or sys.platform == 'riscos':
raise test_support.TestSkipped("Can't test signal on %s" % \
sys.platform)
test_support.run_unittest(BasicSignalTests, InterProcessSignalTests)
if __name__ == "__main__":
test_main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7