How to really test signal handling in Python?

问题:

My code is simple:

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

So I register my application with signal handling of SIGINT, then I start a start a TCP listener.

here are my questions:

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

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

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

回答1:

First of, testing the signal itself is a functional or integration test, not a unit test. See What's the difference between unit, functional, acceptance, and integration tests?

You can run your Python script as a subprocess with subprocess.Popen(), then use the Popen.send_signal() method to send signals to that process, then test that the process has exited with Popen.poll().

回答2:

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

You can use os.kill, which slightly misleadingly, can used to send any signal to any process by its ID. The process ID of the application/test can be found by os.getpid(), so you would have...

 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?

The usual way in a test you can check that some code raises SystemExit, is with 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. If I run start() in my test, it will block and how can I send a signal to it?

This is the trick: you can start another thread which then sends a signal back to the main thread which is blocking.

Putting it all together, assuming your production start function is in 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()

Then your test code could be like the following, say in 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()

and run using

python test.py

---

The above is the same technique as in the answer at https://stackoverflow.com/a/49500820/1319998