python - (单元)测试python信号处理程序

Question

我有一个简单的 Python 服务，其中有一个无限执行某些操作的循环。在各种信号上，sys.exit(0) 被调用，这导致 SystemExit 被引发，然后如果可以的话应该进行一些清理。

在测试中，即标准 unittest.TestCase，我想测试这个清理是否发生并且循环退出。但是，我什至坚持要触发信号/要引发 SystemExit。

# 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处理程序/信号处理程序？也欢迎使用替代模式。

Answer 1

您可以在主线程收到延迟后从另一个线程触发 SIGINT(或任何信号)。然后，您可以像在任何其他测试中一样断言其效果，如下所示。

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()