（单元）测试python信号处理程序

问题描述

我有一个简单的Python服务，其中有一个循环，可以无限执行某些动作。在各种信号上，调用 sys.exit（0），这将引起 SystemExit 的出现，然后进行一些清理

I have a simple Python service, where there is a loop that performs some action infinitely. On various signals, sys.exit(0) is called, which causes SystemExit to be raised and then some cleanup should happen if it can.

在一个测试中，即标准 unittest.TestCase ，我想测试一下清理发生并且循环退出。但是，我仍然坚持要触发信号/发出 SystemExit 。

In a test, i.e. standard unittest.TestCase, I would like to test that this cleanup happens and the loop exits. However, I'm stuck on even getting the signal to be triggered / SystemExit to be raised.

# service.py
import signal
import sys
import time

def main():

    def signal_handler(signalnum, _):
        # How to get this to block to run in a test?
        sys.exit(0)

    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    while True:
        try:
            print("Some action here")
            time.sleep(10)
        except SystemExit:
            # How to get this to block to run in a test?
            print("Some cleanup")
            break

if __name__ == '__main__':
    main()

代码如何在测试环境中输入 SystemExit 处理程序/信号处理程序？

How can the code enter the SystemExit handler / signal handler in the test environment? An alternative pattern would also be welcome.

推荐答案

您可以在一段时间延迟后从另一个线程触发SIGINT（或任何信号），它被接收到主线程中。然后，您可以像在任何其他测试中一样断言其影响，如下所示。

You can trigger a SIGINT (or any signal) from another thread after some delay, which is received in the main thread. You can then assert on its effects just as in any other test, as below.

import os
import signal
import time
import threading
import unittest
from unittest.mock import (
    Mock,
    patch,
)

import service

class TestService(unittest.TestCase):

    @patch('service.print')
    def test_signal_handling(self, mock_print):

        pid = os.getpid()

        def trigger_signal():
            while len(mock_print.mock_calls) < 1:
                time.sleep(0.2)
            os.kill(pid, signal.SIGINT)

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

        service.main()

        self.assertEqual(mock_print.mock_calls[1][1][0], 'Some cleanup')


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

这篇关于（单元）测试python信号处理程序的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！