tests/test_mainloop.py

   1 # -*- Mode: Python -*-
   2
   3 from __future__ import absolute_import
   4
   5 import os
   6 import select
   7 import signal
   8 import unittest
   9
  10 from gi.repository import GLib
  11
  12 from .helper import capture_exceptions
  13
  14
  15 class TestMainLoop(unittest.TestCase):
  16
  17     @unittest.skipUnless(hasattr(os, "fork"), "no os.fork available")
  18     def test_exception_handling(self):
  19         pipe_r, pipe_w = os.pipe()
  20
  21         pid = os.fork()
  22         if pid == 0:
  23             os.close(pipe_w)
  24             select.select([pipe_r], [], [])
  25             os.close(pipe_r)
  26             os._exit(1)
  27
  28         def child_died(pid, status, loop):
  29             loop.quit()
  30             raise Exception("deadbabe")
  31
  32         loop = GLib.MainLoop()
  33         GLib.child_watch_add(GLib.PRIORITY_DEFAULT, pid, child_died, loop)
  34
  35         os.close(pipe_r)
  36         os.write(pipe_w, b"Y")
  37         os.close(pipe_w)
  38
  39         with capture_exceptions() as exc:
  40             loop.run()
  41
  42         assert len(exc) == 1
  43         assert exc[0].type is Exception
  44         assert exc[0].value.args[0] == "deadbabe"
  45
  46     @unittest.skipUnless(hasattr(os, "fork"), "no os.fork available")
  47     @unittest.skipIf(os.environ.get("PYGI_TEST_GDB"), "SIGINT stops gdb")
  48     def test_sigint(self):
  49         r, w = os.pipe()
  50         pid = os.fork()
  51         if pid == 0:
  52             # wait for the parent process loop to start
  53             os.read(r, 1)
  54             os.close(r)
  55
  56             os.kill(os.getppid(), signal.SIGINT)
  57             os._exit(0)
  58
  59         def notify_child():
  60             # tell the child that it can kill the parent
  61             os.write(w, b"X")
  62             os.close(w)
  63
  64         GLib.idle_add(notify_child)
  65         loop = GLib.MainLoop()
  66         try:
  67             loop.run()
  68             self.fail('expected KeyboardInterrupt exception')
  69         except KeyboardInterrupt:
  70             pass
  71         self.assertFalse(loop.is_running())
  72         os.waitpid(pid, 0)