信号具有处理方式、掩码和投递规则

from pathlib import Path
from tempfile import TemporaryDirectory
import errno,os,select,signal,subprocess,sys

watched={signal.SIGUSR1,signal.SIGRTMIN}
old=signal.pthread_sigmask(signal.SIG_BLOCK,watched)
try:
    for _ in range(5):os.kill(os.getpid(),signal.SIGUSR1)
    assert signal.SIGUSR1 in signal.sigpending()
    assert signal.sigwait({signal.SIGUSR1})==signal.SIGUSR1
    assert signal.sigtimedwait({signal.SIGUSR1},0) is None
    for _ in range(3):os.kill(os.getpid(),signal.SIGRTMIN)
    for _ in range(3):assert signal.sigtimedwait({signal.SIGRTMIN},0).si_signo==signal.SIGRTMIN
    assert signal.sigtimedwait({signal.SIGRTMIN},0) is None
    print('five blocked standard signals coalesce into one; three blocked real-time signals remain queued')
finally:signal.pthread_sigmask(signal.SIG_SETMASK,old)
for sig in [signal.SIGKILL,signal.SIGSTOP]:
    try:signal.signal(sig,signal.SIG_IGN)
    except OSError as error:assert error.errno==errno.EINVAL
    else:raise AssertionError('uncatchable signal accepted a disposition change')
old=signal.pthread_sigmask(signal.SIG_BLOCK,{signal.SIGKILL,signal.SIGSTOP})
assert not {signal.SIGKILL,signal.SIGSTOP}&signal.pthread_sigmask(signal.SIG_BLOCK,set())
signal.pthread_sigmask(signal.SIG_SETMASK,old)
print('SIGKILL and SIGSTOP cannot be ignored or added to the blocked mask')
worker=r'''
from pathlib import Path
import signal,sys
if int(sys.argv[1]):signal.signal(signal.SIGTERM,lambda *_:sys.exit(0))
try:
    print('ready',flush=True)
    while True:signal.pause()
finally:Path(sys.argv[2]).write_text('cleanup')
'''
with TemporaryDirectory(prefix='xbcatm-signals-') as directory:
    for name,sig,handled,expected,cleaned in [
        ('hup-default',signal.SIGHUP,False,-signal.SIGHUP,False),
        ('term-default',signal.SIGTERM,False,-signal.SIGTERM,False),
        ('term-handled',signal.SIGTERM,True,0,True),
        ('kill',signal.SIGKILL,True,-signal.SIGKILL,False)]:
        marker=Path(directory)/name
        process=subprocess.Popen([sys.executable,'-c',worker,str(int(handled)),str(marker)],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
        try:
            assert select.select([process.stdout], [], [], 5)[0], 'child readiness timed out'
            assert process.stdout.readline().strip()=='ready'
            os.kill(process.pid,sig);output,error=process.communicate(timeout=5)
            assert process.returncode==expected,(name,process.returncode,error)
            assert marker.exists()==cleaned,name
            print(name,'returncode',process.returncode,'cleanup',cleaned)
        finally:
            if process.poll() is None:process.kill();process.wait()