tests/test_ossig.py

   1 # -*- coding: utf-8 -*-
   2 # Copyright 2017 Christoph Reiter
   3 #
   4 # This library is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU Lesser General Public
   6 # License as published by the Free Software Foundation; either
   7 # version 2.1 of the License, or (at your option) any later version.
   8 #
   9 # This library is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12 # Lesser General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Lesser General Public
  15 # License along with this library; if not, see <http://www.gnu.org/licenses/>.
  16
  17 from __future__ import absolute_import
  18
  19 import os
  20 import signal
  21 import unittest
  22 import threading
  23 from contextlib import contextmanager
  24
  25 try:
  26     from gi.repository import Gtk
  27 except ImportError:
  28     Gtk = None
  29 from gi.repository import Gio, GLib
  30 from gi._ossighelper import wakeup_on_signal, register_sigint_fallback
  31
  32
  33 class TestOverridesWakeupOnAlarm(unittest.TestCase):
  34
  35     @contextmanager
  36     def _run_with_timeout(self, timeout, abort_func):
  37         failed = []
  38
  39         def fail():
  40             abort_func()
  41             failed.append(1)
  42             return True
  43
  44         fail_id = GLib.timeout_add(timeout, fail)
  45         try:
  46             yield
  47         finally:
  48             GLib.source_remove(fail_id)
  49         self.assertFalse(failed)
  50
  51     def test_basic(self):
  52         self.assertEqual(signal.set_wakeup_fd(-1), -1)
  53         with wakeup_on_signal():
  54             pass
  55         self.assertEqual(signal.set_wakeup_fd(-1), -1)
  56
  57     def test_in_thread(self):
  58         failed = []
  59
  60         def target():
  61             try:
  62                 with wakeup_on_signal():
  63                     pass
  64             except:
  65                 failed.append(1)
  66
  67         t = threading.Thread(target=target)
  68         t.start()
  69         t.join(5)
  70         self.assertFalse(failed)
  71
  72     @unittest.skipIf(os.name == "nt", "not on Windows")
  73     def test_glib_mainloop(self):
  74         loop = GLib.MainLoop()
  75         signal.signal(signal.SIGALRM, lambda *args: loop.quit())
  76         GLib.idle_add(signal.setitimer, signal.ITIMER_REAL, 0.001)
  77
  78         with self._run_with_timeout(2000, loop.quit):
  79             loop.run()
  80
  81     @unittest.skipIf(os.name == "nt", "not on Windows")
  82     def test_gio_application(self):
  83         app = Gio.Application()
  84         signal.signal(signal.SIGALRM, lambda *args: app.quit())
  85         GLib.idle_add(signal.setitimer, signal.ITIMER_REAL, 0.001)
  86
  87         with self._run_with_timeout(2000, app.quit):
  88             app.hold()
  89             app.connect("activate", lambda *args: None)
  90             app.run()
  91
  92     @unittest.skipIf(Gtk is None or os.name == "nt", "not on Windows")
  93     def test_gtk_main(self):
  94         signal.signal(signal.SIGALRM, lambda *args: Gtk.main_quit())
  95         GLib.idle_add(signal.setitimer, signal.ITIMER_REAL, 0.001)
  96
  97         with self._run_with_timeout(2000, Gtk.main_quit):
  98             Gtk.main()
  99
 100     @unittest.skipIf(Gtk is None or os.name == "nt", "not on Windows")
 101     def test_gtk_dialog_run(self):
 102         w = Gtk.Window()
 103         d = Gtk.Dialog(transient_for=w)
 104         signal.signal(signal.SIGALRM, lambda *args: d.destroy())
 105         GLib.idle_add(signal.setitimer, signal.ITIMER_REAL, 0.001)
 106
 107         with self._run_with_timeout(2000, d.destroy):
 108             d.run()
 109
 110
 111 class TestSigintFallback(unittest.TestCase):
 112
 113     def setUp(self):
 114         self.assertEqual(
 115             signal.getsignal(signal.SIGINT), signal.default_int_handler)
 116
 117     def tearDown(self):
 118         self.assertEqual(
 119             signal.getsignal(signal.SIGINT), signal.default_int_handler)
 120
 121     def test_replace_handler_and_restore_nested(self):
 122         with register_sigint_fallback(lambda: None):
 123             new_handler = signal.getsignal(signal.SIGINT)
 124             self.assertNotEqual(new_handler, signal.default_int_handler)
 125             with register_sigint_fallback(lambda: None):
 126                 self.assertTrue(signal.getsignal(signal.SIGINT) is new_handler)
 127         self.assertEqual(
 128             signal.getsignal(signal.SIGINT), signal.default_int_handler)
 129
 130     def test_no_replace_if_not_default(self):
 131         new_handler = lambda *args: None
 132         signal.signal(signal.SIGINT, new_handler)
 133         try:
 134             with register_sigint_fallback(lambda: None):
 135                 self.assertTrue(signal.getsignal(signal.SIGINT) is new_handler)
 136                 with register_sigint_fallback(lambda: None):
 137                     self.assertTrue(
 138                         signal.getsignal(signal.SIGINT) is new_handler)
 139             self.assertTrue(signal.getsignal(signal.SIGINT) is new_handler)
 140         finally:
 141             signal.signal(signal.SIGINT, signal.default_int_handler)
 142
 143     def test_noop_in_threads(self):
 144         failed = []
 145
 146         def target():
 147             try:
 148                 with register_sigint_fallback(lambda: None):
 149                     with register_sigint_fallback(lambda: None):
 150                         self.assertTrue(
 151                             signal.getsignal(signal.SIGINT) is
 152                             signal.default_int_handler)
 153             except:
 154                 failed.append(1)
 155
 156         t = threading.Thread(target=target)
 157         t.start()
 158         t.join(5)
 159         self.assertFalse(failed)
 160
 161     @unittest.skipIf(os.name == "nt", "not on Windows")
 162     def test_no_replace_if_set_by_glib(self):
 163         id_ = GLib.unix_signal_add(
 164             GLib.PRIORITY_DEFAULT, signal.SIGINT, lambda *args: None)
 165         try:
 166             # signal.getsignal() doesn't pick up that unix_signal_add()
 167             # has changed the handler, but we should anyway.
 168             self.assertEqual(
 169                 signal.getsignal(signal.SIGINT), signal.default_int_handler)
 170             with register_sigint_fallback(lambda: None):
 171                 self.assertEqual(
 172                     signal.getsignal(signal.SIGINT),
 173                     signal.default_int_handler)
 174             self.assertEqual(
 175                 signal.getsignal(signal.SIGINT), signal.default_int_handler)
 176         finally:
 177             GLib.source_remove(id_)
 178             signal.signal(signal.SIGINT, signal.SIG_DFL)
 179             signal.signal(signal.SIGINT, signal.default_int_handler)