如何在Python中真正测试信号处理？

Question

我的代码很简单:

def start():
    signal(SIGINT, lambda signal, frame: raise SystemExit())
    startTCPServer()

所以我用信号处理注册我的应用程序SIGINT,然后我开始启动一个TCP监听器.

这是我的问题:

1. 如何使用python代码发送SIGINT信号？

2. 如何测试应用程序是否收到SIGINT信号,是否会引发SystemExit异常？

3. 如果我start()在测试中运行,它将阻止,我该如何向其发送信号？

Answer 1

1. 如何使用 python 代码发送 SIGINT 信号？

您可以使用os.kill，这有点误导，可用于通过 ID 向任何进程发送任何信号。应用程序/测试的进程 ID 可以通过 找到os.getpid()，因此您可以...

 pid = os.getpid()
 #  ... other code discussed later in the answer ...
 os.kill(pid, SIGINT)

2. 如何测试应用程序收到 SIGINT 信号是否会引发 SystemExit 异常？

在测试中，您可以检查某些代码是否引发 SystemExit 的常用方法是unittest.TestCase::assertRaises...

import start

class TestStart(unittest.TestCase):

    def test_signal_handling(self):

        #  ... other code discussed later in the answer ...

        with self.assertRaises(SystemExit):
           start.start()

3. 如果我在测试中运行 start() ，它将阻塞，我如何向它发送信号？

这就是技巧：您可以启动另一个线程，然后该线程将信号发送回阻塞的主线程。

将它们放在一起，假设您的生产start函数位于start.py：

from signal import (
    SIGINT,
    signal,
)
import socketserver

def startTCPServer():
    # Taken from https://docs.python.org/3.4/library/socketserver.html#socketserver-tcpserver-example
    class MyTCPHandler(socketserver.BaseRequestHandler):
        def handle(self):
            self.data = self.request.recv(1024).strip()
            self.request.sendall(self.data.upper())

    HOST, PORT = "localhost", 9999
    server = socketserver.TCPServer((HOST, PORT), MyTCPHandler)
    server.serve_forever()

def start():
    def raiseSystemExit(_, __):
        raise SystemExit

    signal(SIGINT, raiseSystemExit)
    startTCPServer()

那么你的测试代码可能如下所示test.py

import os
from signal import (
    SIGINT,
)
import threading
import time
import unittest

import start


class TestStart(unittest.TestCase):

    def test_signal_handling(self):
        pid = os.getpid()

        def trigger_signal():
            # You could do something more robust, e.g. wait until port is listening
            time.sleep(1)
            os.kill(pid, SIGINT)

        thread = threading.Thread(target=trigger_signal)
        thread.daemon = True
        thread.start()

        with self.assertRaises(SystemExit):
            start.start()


if __name__ == '__main__':
    unittest.main()

并使用运行

python test.py

以上与/sf/answers/3465057431/的答案中的技术相同

Answer 2

首先，测试信号本身是功能测试或集成测试，而不是单元测试。请参阅单元，功能，验收和集成测试之间的区别是什么？

您可以subprocess.Popen()使用作为子进程运行Python脚本，然后使用Popen.send_signal()方法将信号发送到该进程，然后使用来测试该进程是否已退出Popen.poll()。