如何从单元测试模拟终端 CTRL + C 事件?

How to simulate a terminal CTRL + C event from a unittest?

我有一个 multiprocessing.Process 子类忽略了 SIGINT:

# inside the run method
signal.signal(signal.SIGINT, signal.SIG_IGN)

我不希望这个进程在按下 CTRL + C 时终止,所以我试图模拟这个终端事件在我的单元测试中,通过向此进程 ID 发送 SIGINT 信号:

os.kill(PID, signal.SIGINT)

但是即使不忽略这个信号,进程也不会终止,所以这个测试是没用的,我从其他问题中发现在 CTRL + C 终端向进程组 ID 发送 SIGINT 事件,但我不能这样做,因为它也会终止单元测试进程。

那么为什么进程在收到来自 os.killSIGINT 时没有终止?我应该用另一种方式来做吗?

child 进程应在收到 SIGINT 后终止,除非它忽略该信号或安装了自己的处理程序。如果您没有在 child 中明确忽略 SIGINT,那么 SIGINT 可能在 parent 中被忽略,因此在 child 中,因为信号配置是继承的。

但是,我无法重现您的问题,事实上,我发现了相反的问题:child 进程 终止 无论其信号处置如何。

如果信号发送得太快,在child进程忽略SIGINT(在它的run()方法中)之前,它将被终止。下面是一些演示问题的代码:

import os, time, signal
from multiprocessing import Process

class P(Process):
    def run(self):
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        return super(P, self).run()

def f():
    print 'Child sleeping...'
    time.sleep(10)
    print 'Child done'

p = P(target=f)
p.start()    
print 'Child started with PID', p.pid

print 'Killing child'
os.kill(p.pid, signal.SIGINT)
print 'Joining child'
p.join()

输出

Child started with PID 1515
Killing child
Joining child
Traceback (most recent call last):
  File "p1.py", line 15, in 
    p.start()    
  File "/usr/lib64/python2.7/multiprocessing/process.py", line 130, in start
    self._popen = Popen(self)
  File "/usr/lib64/python2.7/multiprocessing/forking.py", line 126, in __init__
    code = process_obj._bootstrap()
  File "/usr/lib64/python2.7/multiprocessing/process.py", line 242, in _bootstrap
    from . import util
KeyboardInterrupt

在将 SIGINT 信号发送到 child 之前,在 parent 中使用 time.sleep(0.1) 添加一个小的延迟将解决问题。这将为 child 提供足够的时间来执行忽略 SIGINT 的 run() 方法。现在信号将被 child:

忽略
Child started with PID 1589
Killing child
Child sleeping...
Joining child
Child done

不需要延迟也不需要自定义 run() 方法的替代方法是将 parent 设置为忽略 SIGINT,启动 child,然后恢复 parent 的原始 SIGINT 处理程序。因为信号配置是继承的,所以 child 从它开始的那一刻起就会忽略 SIGINT:

import os, time, signal
from multiprocessing import Process

def f():
    print 'Child sleeping...'
    time.sleep(10)
    print 'Child done'

p = Process(target=f)
old_sigint = signal.signal(signal.SIGINT, signal.SIG_IGN)
p.start()    
signal.signal(signal.SIGINT, old_sigint)    # restore parent's handler
print 'Child started with PID', p.pid

print 'Killing child'
os.kill(p.pid, signal.SIGINT)
print 'Joining child'
p.join()

输出

Child started with PID 1660
Killing child
Joining child
Child sleeping...
Child done

问题的简化版本是:

import os, time, signal

childpid = os.fork()

if childpid == 0:
    # in the child
    time.sleep(5)      # will be interrupted by KeyboardInterrupt
    print "stop child"
else:
    # in the parent
    #time.sleep(1)
    os.kill(childpid, signal.SIGINT)

如果 parent 在发送信号之前执行 sleep(1),则一切都按预期工作:child(并且只有 child)收到 Python KeyboardInterrupt异常,即中断sleep(5)。然而,如果我们像上面的例子一样注释掉 sleep(1)kill() 似乎被完全忽略:child 运行,休眠 5 秒,最后打印 "stop child"。因此,您的测试套件可能有一个简单的解决方法:只需添加一个小的 sleep().

据我所知,发生这种情况的原因如下(坏的):查看 CPython 源代码,在系统调用 fork() 之后,child 进程明确清除挂起信号列表。但是下面的情况似乎经常发生:parent继续稍微领先于child,并发送SIGINT信号。 child 收到它,但此时它仍然只是在系统调用 fork() 之后不久,在 _clear_pending_signals() 之前。结果,信号丢失了。

如果您想在 http://bugs.python.org 上提交问题,这可以被视为 CPython 错误。请参阅 signalmodule.c 中的 PyOS_AfterFork()。