(Unidad) Controlador de señales de python de prueba

Tengo un servicio Python simple, donde hay un bucle que realiza alguna acción infinitamente. En varias señales, se sys.exit(0) , lo que hace que SystemExit y luego, si es posible, debe realizarse una limpieza.

En una prueba, es decir, unittest.TestCase estándar, me gustaría probar que esta limpieza se realiza y el bucle sale. Sin embargo, estoy atascado en incluso obtener la señal para ser activada / SystemExit para ser levantada.

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

¿Cómo puede el código ingresar al controlador SystemExit / controlador de señal en el entorno de prueba? Un patrón alternativo también sería bienvenido.

Vamos a refactorizar eso para que sea más fácil de probar:

 def loop(): try: print("Some action here") except: # clean up and re-raise print("Some cleanup") raise 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: loop_body() time.sleep(10) except SystemExit: break if __name__ == '__main__': main() 

Sin embargo, esto no permite una prueba fácil del código de manejo de la señal. Sin embargo, esa cantidad es tan pequeña que rara vez cambia y depende en gran medida del entorno, por lo que es posible y quizás incluso mejor probarla manualmente.

Para mayor claridad, podría ser útil usar un controlador de contexto, lo que suele ser una buena idea cuando tiene un código de configuración / apagado. No menciona el código de configuración, pero mi Bola de Cristal ™ me dice que existe. Entonces podría llamarse así:

 try: with my_service() as service: while True: service.run() sleep(10) except SystemExit: # perform graceful shutdown on signal pass 

Dejaré la implementación de ese administrador de contexto para ti, pero echa un vistazo a contextlib , lo que lo hace fácil y divertido.

Puede activar un SIGINT (o cualquier señal) desde otro hilo después de algún retraso, que se recibe en el hilo principal. A continuación, puede hacer valer sus efectos como en cualquier otra prueba, como se muestra a continuación.

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