summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_signal.py
blob: 737f1608c9f873a633a708289b994498aeb97d27 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
# Test the signal module
from test.test_support import verbose, TestSkipped, TestFailed
import signal
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

if verbose:
    x = '-x'
else:
    x = '+x'
pid = os.getpid()

# Shell script that will send us asynchronous signals
script = """
 (
        set %(x)s
        sleep 2
        kill -5 %(pid)d
        sleep 2
        kill -2 %(pid)d
        sleep 2
        kill -3 %(pid)d
 ) &
""" % vars()

def handlerA(*args):
    if verbose:
        print "handlerA", args

HandlerBCalled = "HandlerBCalled"       # Exception

def handlerB(*args):
    if verbose:
        print "handlerB", args
    raise HandlerBCalled, args

signal.alarm(20)                        # Entire test lasts at most 20 sec.
signal.signal(5, handlerA)
signal.signal(2, handlerB)
signal.signal(3, signal.SIG_IGN)
signal.signal(signal.SIGALRM, signal.default_int_handler)

os.system(script)

print "starting pause() loop..."

try:
    while 1:
        if verbose:
            print "call pause()..."
        try:
            signal.pause()
            if verbose:
                print "pause() returned"
        except HandlerBCalled:
            if verbose:
                print "HandlerBCalled exception caught"
            else:
                pass

except KeyboardInterrupt:
    if verbose:
        print "KeyboardInterrupt (assume the alarm() went off)"


if hasattr(signal, "sigprocmask"):
    class HupDelivered(Exception):
        pass
    def hup(signum, frame):
        raise HupDelivered
    def hup2(signum, frame):
        signal.signal(signal.SIGHUP, hup)
        return
    signal.signal(signal.SIGHUP, hup)

    if verbose:
        print "blocking SIGHUP"

    defaultmask = signal.sigprocmask(signal.SIG_BLOCK, [signal.SIGHUP])

    if verbose:
        print "sending SIGHUP"

    try:
        os.kill(pid, signal.SIGHUP)
    except HupDelivered:
        raise TestFailed, "HUP not blocked"

    if signal.SIGHUP not in signal.sigpending():
        raise TestFailed, "HUP not pending"

    if verbose:
        print "unblocking SIGHUP"

    try:
        signal.sigprocmask(signal.SIG_UNBLOCK, [signal.SIGHUP])
    except HupDelivered:
        pass
    else:
        raise TestFailed, "HUP not delivered"

    if verbose:
        print "testing sigsuspend"

    signal.sigprocmask(signal.SIG_BLOCK, [signal.SIGHUP])
    signal.signal(signal.SIGHUP, hup2)

    if not os.fork():
        time.sleep(2)
        os.kill(pid, signal.SIGHUP)
        time.sleep(2)
        os.kill(pid, signal.SIGHUP)
        os._exit(0)
    else:
        try:
            signal.sigsuspend(defaultmask)
        except:
            raise TestFailed, "sigsuspend erroneously raised"

        try:
            signal.sigsuspend(defaultmask)
        except HupDelivered:
            pass
        else:
            raise TestFailed, "sigsupsend didn't raise"