twisted/internet/test/test_sigchld.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for L{twisted.internet._sigchld}, an alternate, superior SIGCHLD
   6 monitoring API.
   7 """
   8
   9 import os, signal, errno
  10
  11 from twisted.python.log import msg
  12 from twisted.trial.unittest import TestCase
  13 from twisted.internet.fdesc import setNonBlocking
  14 from twisted.internet._signals import installHandler, isDefaultHandler
  15 from twisted.internet._signals import _extInstallHandler, _extIsDefaultHandler
  16 from twisted.internet._signals import _installHandlerUsingSetWakeup, \
  17     _installHandlerUsingSignal, _isDefaultHandler
  18
  19
  20 class SIGCHLDTestsMixin:
  21     """
  22     Mixin for L{TestCase} subclasses which defines several tests for
  23     I{installHandler} and I{isDefaultHandler}.  Subclasses are expected to
  24     define C{self.installHandler} and C{self.isDefaultHandler} to invoke the
  25     implementation to be tested.
  26     """
  27
  28     if getattr(signal, 'SIGCHLD', None) is None:
  29         skip = "Platform does not have SIGCHLD"
  30
  31     def installHandler(self, fd):
  32         """
  33         Override in a subclass to install a SIGCHLD handler which writes a byte
  34         to the given file descriptor.  Return the previously registered file
  35         descriptor.
  36         """
  37         raise NotImplementedError()
  38
  39
  40     def isDefaultHandler(self):
  41         """
  42         Override in a subclass to determine if the current SIGCHLD handler is
  43         SIG_DFL or not.  Return True if it is SIG_DFL, False otherwise.
  44         """
  45         raise NotImplementedError()
  46
  47
  48     def pipe(self):
  49         """
  50         Create a non-blocking pipe which will be closed after the currently
  51         running test.
  52         """
  53         read, write = os.pipe()
  54         self.addCleanup(os.close, read)
  55         self.addCleanup(os.close, write)
  56         setNonBlocking(read)
  57         setNonBlocking(write)
  58         return read, write
  59
  60
  61     def setUp(self):
  62         """
  63         Save the current SIGCHLD handler as reported by L{signal.signal} and
  64         the current file descriptor registered with L{installHandler}.
  65         """
  66         handler = signal.getsignal(signal.SIGCHLD)
  67         if handler != signal.SIG_DFL:
  68             self.signalModuleHandler = handler
  69             signal.signal(signal.SIGCHLD, signal.SIG_DFL)
  70         else:
  71             self.signalModuleHandler = None
  72
  73         self.oldFD = self.installHandler(-1)
  74
  75         if self.signalModuleHandler is not None and self.oldFD != -1:
  76             msg("SIGCHLD setup issue: %r %r" % (self.signalModuleHandler, self.oldFD))
  77             raise RuntimeError("You used some signal APIs wrong!  Try again.")
  78
  79
  80     def tearDown(self):
  81         """
  82         Restore whatever signal handler was present when setUp ran.
  83         """
  84         # If tests set up any kind of handlers, clear them out.
  85         self.installHandler(-1)
  86         signal.signal(signal.SIGCHLD, signal.SIG_DFL)
  87
  88         # Now restore whatever the setup was before the test ran.
  89         if self.signalModuleHandler is not None:
  90             signal.signal(signal.SIGCHLD, self.signalModuleHandler)
  91         elif self.oldFD != -1:
  92             self.installHandler(self.oldFD)
  93
  94
  95     def test_isDefaultHandler(self):
  96         """
  97         L{isDefaultHandler} returns true if the SIGCHLD handler is SIG_DFL,
  98         false otherwise.
  99         """
 100         self.assertTrue(self.isDefaultHandler())
 101         signal.signal(signal.SIGCHLD, signal.SIG_IGN)
 102         self.assertFalse(self.isDefaultHandler())
 103         signal.signal(signal.SIGCHLD, signal.SIG_DFL)
 104         self.assertTrue(self.isDefaultHandler())
 105         signal.signal(signal.SIGCHLD, lambda *args: None)
 106         self.assertFalse(self.isDefaultHandler())
 107
 108
 109     def test_returnOldFD(self):
 110         """
 111         L{installHandler} returns the previously registered file descriptor.
 112         """
 113         read, write = self.pipe()
 114         oldFD = self.installHandler(write)
 115         self.assertEqual(self.installHandler(oldFD), write)
 116
 117
 118     def test_uninstallHandler(self):
 119         """
 120         C{installHandler(-1)} removes the SIGCHLD handler completely.
 121         """
 122         read, write = self.pipe()
 123         self.assertTrue(self.isDefaultHandler())
 124         self.installHandler(write)
 125         self.assertFalse(self.isDefaultHandler())
 126         self.installHandler(-1)
 127         self.assertTrue(self.isDefaultHandler())
 128
 129
 130     def test_installHandler(self):
 131         """
 132         The file descriptor passed to L{installHandler} has a byte written to
 133         it when SIGCHLD is delivered to the process.
 134         """
 135         read, write = self.pipe()
 136         self.installHandler(write)
 137
 138         exc = self.assertRaises(OSError, os.read, read, 1)
 139         self.assertEqual(exc.errno, errno.EAGAIN)
 140
 141         os.kill(os.getpid(), signal.SIGCHLD)
 142
 143         self.assertEqual(len(os.read(read, 5)), 1)
 144
 145
 146
 147 class DefaultSIGCHLDTests(SIGCHLDTestsMixin, TestCase):
 148     """
 149     Tests for whatever implementation is selected for the L{installHandler}
 150     and L{isDefaultHandler} APIs.
 151     """
 152     installHandler = staticmethod(installHandler)
 153     isDefaultHandler = staticmethod(isDefaultHandler)
 154
 155
 156
 157 class ExtensionSIGCHLDTests(SIGCHLDTestsMixin, TestCase):
 158     """
 159     Tests for the L{twisted.internet._sigchld} implementation of the
 160     L{installHandler} and L{isDefaultHandler} APIs.
 161     """
 162     try:
 163         import twisted.internet._sigchld
 164     except ImportError:
 165         skip = "twisted.internet._sigchld is not available"
 166
 167     installHandler = _extInstallHandler
 168     isDefaultHandler = _extIsDefaultHandler
 169
 170
 171
 172 class SetWakeupSIGCHLDTests(SIGCHLDTestsMixin, TestCase):
 173     """
 174     Tests for the L{signal.set_wakeup_fd} implementation of the
 175     L{installHandler} and L{isDefaultHandler} APIs.
 176     """
 177     # Check both of these.  On Ubuntu 9.10 (to take an example completely at
 178     # random), Python 2.5 has set_wakeup_fd but not siginterrupt.
 179     if (getattr(signal, 'set_wakeup_fd', None) is None
 180         or getattr(signal, 'siginterrupt', None) is None):
 181         skip = "signal.set_wakeup_fd is not available"
 182
 183     installHandler = staticmethod(_installHandlerUsingSetWakeup)
 184     isDefaultHandler = staticmethod(_isDefaultHandler)
 185
 186
 187
 188 class PlainSignalModuleSIGCHLDTests(SIGCHLDTestsMixin, TestCase):
 189     """
 190     Tests for the L{signal.signal} implementation of the L{installHandler}
 191     and L{isDefaultHandler} APIs.
 192     """
 193     installHandler = staticmethod(_installHandlerUsingSignal)
 194     isDefaultHandler = staticmethod(_isDefaultHandler)