tests/test_mainloop.py

   1 # -*- Mode: Python -*-
   2
   3 import os
   4 import sys
   5 import select
   6 import signal
   7 import time
   8 import unittest
   9
  10 try:
  11     from _thread import start_new_thread
  12     start_new_thread  # pyflakes
  13 except ImportError:
  14     # Python 2
  15     from thread import start_new_thread
  16 from gi.repository import GLib
  17
  18 from compathelper import _bytes
  19
  20
  21 class TestMainLoop(unittest.TestCase):
  22     def test_exception_handling(self):
  23         pipe_r, pipe_w = os.pipe()
  24
  25         pid = os.fork()
  26         if pid == 0:
  27             os.close(pipe_w)
  28             select.select([pipe_r], [], [])
  29             os.close(pipe_r)
  30             os._exit(1)
  31
  32         def child_died(pid, status, loop):
  33             loop.quit()
  34             raise Exception("deadbabe")
  35
  36         loop = GLib.MainLoop()
  37         GLib.child_watch_add(GLib.PRIORITY_DEFAULT, pid, child_died, loop)
  38
  39         os.close(pipe_r)
  40         os.write(pipe_w, _bytes("Y"))
  41         os.close(pipe_w)
  42
  43         def excepthook(type, value, traceback):
  44             self.assertTrue(type is Exception)
  45             self.assertEqual(value.args[0], "deadbabe")
  46         sys.excepthook = excepthook
  47         try:
  48             got_exception = False
  49             try:
  50                 loop.run()
  51             except:
  52                 got_exception = True
  53         finally:
  54             sys.excepthook = sys.__excepthook__
  55
  56         #
  57         # The exception should be handled (by printing it)
  58         # immediately on return from child_died() rather
  59         # than here. See bug #303573
  60         #
  61         self.assertFalse(got_exception)
  62
  63     def test_concurrency(self):
  64         def on_usr1(signum, frame):
  65             pass
  66
  67         try:
  68             # create a thread which will terminate upon SIGUSR1 by way of
  69             # interrupting sleep()
  70             orig_handler = signal.signal(signal.SIGUSR1, on_usr1)
  71             start_new_thread(time.sleep, (10,))
  72
  73             # now create two main loops
  74             loop1 = GLib.MainLoop()
  75             loop2 = GLib.MainLoop()
  76             GLib.timeout_add(100, lambda: os.kill(os.getpid(), signal.SIGUSR1))
  77             GLib.timeout_add(500, loop1.quit)
  78             loop1.run()
  79             loop2.quit()
  80         finally:
  81             signal.signal(signal.SIGUSR1, orig_handler)
  82
  83     def test_sigint(self):
  84         pid = os.fork()
  85         if pid == 0:
  86             time.sleep(0.5)
  87             os.kill(os.getppid(), signal.SIGINT)
  88             os._exit(0)
  89
  90         loop = GLib.MainLoop()
  91         try:
  92             loop.run()
  93             self.fail('expected KeyboardInterrupt exception')
  94         except KeyboardInterrupt:
  95             pass
  96         self.assertFalse(loop.is_running())
  97         os.waitpid(pid, 0)