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

Question

我的代码很简单：

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

所以我用SIGINT的信号处理注册我的应用程序，然后启动一个启动TCP侦听器。在

以下是我的问题：

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

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

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

Answer 1

首先，测试信号本身是功能测试或集成测试，而不是单元测试。见What's the difference between unit, functional, acceptance, and integration tests?

可以使用^{}作为子进程运行Python脚本，然后使用^{} method向该进程发送信号，然后测试该进程是否已使用^{}退出。在

Answer 2

1. How can I using python code to send a SIGINT signal?

您可以使用os.kill，这有点误导性，它可以用来通过它的ID向任何进程发送任何信号。。。在

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

2. How can I test whether if the application receives a signal of SIGINT, it will raise a SystemExit exception?

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

^{pr2}$

3. If I run start() in my test, it will block and how can I send a signal to it?

这就是诀窍：你可以启动另一个线程，然后将一个信号发送回阻塞的主线程。在

把它们放在一起，假设你的生产函数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

---

上面的技巧与https://stackoverflow.com/a/49500820/1319998中的答案相同