|
1 # Test the signal module |
|
2 from test.test_support import verbose, TestSkipped, TestFailed, vereq |
|
3 import signal |
|
4 import os, sys, time |
|
5 |
|
6 if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos': |
|
7 raise TestSkipped, "Can't test signal on %s" % sys.platform |
|
8 |
|
9 MAX_DURATION = 20 # Entire test should last at most 20 sec. |
|
10 |
|
11 if verbose: |
|
12 x = '-x' |
|
13 else: |
|
14 x = '+x' |
|
15 |
|
16 pid = os.getpid() |
|
17 if verbose: |
|
18 print "test runner's pid is", pid |
|
19 |
|
20 # Shell script that will send us asynchronous signals |
|
21 script = """ |
|
22 ( |
|
23 set %(x)s |
|
24 sleep 2 |
|
25 kill -HUP %(pid)d |
|
26 sleep 2 |
|
27 kill -USR1 %(pid)d |
|
28 sleep 2 |
|
29 kill -USR2 %(pid)d |
|
30 ) & |
|
31 """ % vars() |
|
32 |
|
33 a_called = b_called = False |
|
34 |
|
35 def handlerA(*args): |
|
36 global a_called |
|
37 a_called = True |
|
38 if verbose: |
|
39 print "handlerA invoked", args |
|
40 |
|
41 class HandlerBCalled(Exception): |
|
42 pass |
|
43 |
|
44 def handlerB(*args): |
|
45 global b_called |
|
46 b_called = True |
|
47 if verbose: |
|
48 print "handlerB invoked", args |
|
49 raise HandlerBCalled, args |
|
50 |
|
51 # Set up a child to send signals to us (the parent) after waiting long |
|
52 # enough to receive the alarm. It seems we miss the alarm for some |
|
53 # reason. This will hopefully stop the hangs on Tru64/Alpha. |
|
54 # Alas, it doesn't. Tru64 appears to miss all the signals at times, or |
|
55 # seemingly random subsets of them, and nothing done in force_test_exit |
|
56 # so far has actually helped. |
|
57 def force_test_exit(): |
|
58 # Sigh, both imports seem necessary to avoid errors. |
|
59 import os |
|
60 fork_pid = os.fork() |
|
61 if fork_pid: |
|
62 # In parent. |
|
63 return fork_pid |
|
64 |
|
65 # In child. |
|
66 import os, time |
|
67 try: |
|
68 # Wait 5 seconds longer than the expected alarm to give enough |
|
69 # time for the normal sequence of events to occur. This is |
|
70 # just a stop-gap to try to prevent the test from hanging. |
|
71 time.sleep(MAX_DURATION + 5) |
|
72 print >> sys.__stdout__, ' child should not have to kill parent' |
|
73 for signame in "SIGHUP", "SIGUSR1", "SIGUSR2", "SIGALRM": |
|
74 os.kill(pid, getattr(signal, signame)) |
|
75 print >> sys.__stdout__, " child sent", signame, "to", pid |
|
76 time.sleep(1) |
|
77 finally: |
|
78 os._exit(0) |
|
79 |
|
80 # Install handlers. |
|
81 hup = signal.signal(signal.SIGHUP, handlerA) |
|
82 usr1 = signal.signal(signal.SIGUSR1, handlerB) |
|
83 usr2 = signal.signal(signal.SIGUSR2, signal.SIG_IGN) |
|
84 alrm = signal.signal(signal.SIGALRM, signal.default_int_handler) |
|
85 |
|
86 try: |
|
87 |
|
88 signal.alarm(MAX_DURATION) |
|
89 vereq(signal.getsignal(signal.SIGHUP), handlerA) |
|
90 vereq(signal.getsignal(signal.SIGUSR1), handlerB) |
|
91 vereq(signal.getsignal(signal.SIGUSR2), signal.SIG_IGN) |
|
92 vereq(signal.getsignal(signal.SIGALRM), signal.default_int_handler) |
|
93 |
|
94 # Try to ensure this test exits even if there is some problem with alarm. |
|
95 # Tru64/Alpha often hangs and is ultimately killed by the buildbot. |
|
96 fork_pid = force_test_exit() |
|
97 |
|
98 try: |
|
99 signal.getsignal(4242) |
|
100 raise TestFailed('expected ValueError for invalid signal # to ' |
|
101 'getsignal()') |
|
102 except ValueError: |
|
103 pass |
|
104 |
|
105 try: |
|
106 signal.signal(4242, handlerB) |
|
107 raise TestFailed('expected ValueError for invalid signal # to ' |
|
108 'signal()') |
|
109 except ValueError: |
|
110 pass |
|
111 |
|
112 try: |
|
113 signal.signal(signal.SIGUSR1, None) |
|
114 raise TestFailed('expected TypeError for non-callable') |
|
115 except TypeError: |
|
116 pass |
|
117 |
|
118 # Launch an external script to send us signals. |
|
119 # We expect the external script to: |
|
120 # send HUP, which invokes handlerA to set a_called |
|
121 # send USR1, which invokes handlerB to set b_called and raise |
|
122 # HandlerBCalled |
|
123 # send USR2, which is ignored |
|
124 # |
|
125 # Then we expect the alarm to go off, and its handler raises |
|
126 # KeyboardInterrupt, finally getting us out of the loop. |
|
127 os.system(script) |
|
128 try: |
|
129 print "starting pause() loop..." |
|
130 while 1: |
|
131 try: |
|
132 if verbose: |
|
133 print "call pause()..." |
|
134 signal.pause() |
|
135 if verbose: |
|
136 print "pause() returned" |
|
137 except HandlerBCalled: |
|
138 if verbose: |
|
139 print "HandlerBCalled exception caught" |
|
140 |
|
141 except KeyboardInterrupt: |
|
142 if verbose: |
|
143 print "KeyboardInterrupt (the alarm() went off)" |
|
144 |
|
145 if not a_called: |
|
146 print 'HandlerA not called' |
|
147 |
|
148 if not b_called: |
|
149 print 'HandlerB not called' |
|
150 |
|
151 finally: |
|
152 # Forcibly kill the child we created to ping us if there was a test error. |
|
153 try: |
|
154 # Make sure we don't kill ourself if there was a fork error. |
|
155 if fork_pid > 0: |
|
156 os.kill(fork_pid, signal.SIGKILL) |
|
157 except: |
|
158 # If the child killed us, it has probably exited. Killing a |
|
159 # non-existent process will raise an error which we don't care about. |
|
160 pass |
|
161 |
|
162 # Restore handlers. |
|
163 signal.alarm(0) # cancel alarm in case we died early |
|
164 signal.signal(signal.SIGHUP, hup) |
|
165 signal.signal(signal.SIGUSR1, usr1) |
|
166 signal.signal(signal.SIGUSR2, usr2) |
|
167 signal.signal(signal.SIGALRM, alrm) |