signalinterproctester.py 2.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. import os
  2. import signal
  3. import subprocess
  4. import sys
  5. import time
  6. import unittest
  7. from test import support
  8. class SIGUSR1Exception(Exception):
  9. pass
  10. class InterProcessSignalTests(unittest.TestCase):
  11. def setUp(self):
  12. self.got_signals = {'SIGHUP': 0, 'SIGUSR1': 0, 'SIGALRM': 0}
  13. def sighup_handler(self, signum, frame):
  14. self.got_signals['SIGHUP'] += 1
  15. def sigusr1_handler(self, signum, frame):
  16. self.got_signals['SIGUSR1'] += 1
  17. raise SIGUSR1Exception
  18. def wait_signal(self, child, signame):
  19. if child is not None:
  20. # This wait should be interrupted by exc_class
  21. # (if set)
  22. child.wait()
  23. timeout = support.SHORT_TIMEOUT
  24. deadline = time.monotonic() + timeout
  25. while time.monotonic() < deadline:
  26. if self.got_signals[signame]:
  27. return
  28. signal.pause()
  29. self.fail('signal %s not received after %s seconds'
  30. % (signame, timeout))
  31. def subprocess_send_signal(self, pid, signame):
  32. code = 'import os, signal; os.kill(%s, signal.%s)' % (pid, signame)
  33. args = [sys.executable, '-I', '-c', code]
  34. return subprocess.Popen(args)
  35. def test_interprocess_signal(self):
  36. # Install handlers. This function runs in a sub-process, so we
  37. # don't worry about re-setting the default handlers.
  38. signal.signal(signal.SIGHUP, self.sighup_handler)
  39. signal.signal(signal.SIGUSR1, self.sigusr1_handler)
  40. signal.signal(signal.SIGUSR2, signal.SIG_IGN)
  41. signal.signal(signal.SIGALRM, signal.default_int_handler)
  42. # Let the sub-processes know who to send signals to.
  43. pid = str(os.getpid())
  44. with self.subprocess_send_signal(pid, "SIGHUP") as child:
  45. self.wait_signal(child, 'SIGHUP')
  46. self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 0,
  47. 'SIGALRM': 0})
  48. with self.assertRaises(SIGUSR1Exception):
  49. with self.subprocess_send_signal(pid, "SIGUSR1") as child:
  50. self.wait_signal(child, 'SIGUSR1')
  51. self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
  52. 'SIGALRM': 0})
  53. with self.subprocess_send_signal(pid, "SIGUSR2") as child:
  54. # Nothing should happen: SIGUSR2 is ignored
  55. child.wait()
  56. try:
  57. with self.assertRaises(KeyboardInterrupt):
  58. signal.alarm(1)
  59. self.wait_signal(None, 'SIGALRM')
  60. self.assertEqual(self.got_signals, {'SIGHUP': 1, 'SIGUSR1': 1,
  61. 'SIGALRM': 0})
  62. finally:
  63. signal.alarm(0)
  64. if __name__ == "__main__":
  65. unittest.main()