Back to index

python3.2  3.2.2
Public Member Functions | Public Attributes | Static Public Attributes
test.test_signal.InterProcessSignalTests Class Reference

List of all members.

Public Member Functions

def setUp
def tearDown
def format_frame
def handlerA
def handlerB
def wait
def run_test
def test_main

Public Attributes

 using_gc
 a_called
 b_called

Static Public Attributes

int MAX_DURATION = 20

Detailed Description

Definition at line 44 of file test_signal.py.


Member Function Documentation

def test.test_signal.InterProcessSignalTests.format_frame (   self,
  frame,
  limit = None 
)

Definition at line 55 of file test_signal.py.

00055 
00056     def format_frame(self, frame, limit=None):
00057         return ''.join(traceback.format_stack(frame, limit=limit))

Here is the call graph for this function:

def test.test_signal.InterProcessSignalTests.handlerA (   self,
  signum,
  frame 
)

Definition at line 58 of file test_signal.py.

00058 
00059     def handlerA(self, signum, frame):
00060         self.a_called = True
00061         if support.verbose:
00062             print("handlerA invoked from signal %s at:\n%s" % (
00063                 signum, self.format_frame(frame, limit=1)))

Here is the caller graph for this function:

def test.test_signal.InterProcessSignalTests.handlerB (   self,
  signum,
  frame 
)

Definition at line 64 of file test_signal.py.

00064 
00065     def handlerB(self, signum, frame):
00066         self.b_called = True
00067         if support.verbose:
00068             print ("handlerB invoked from signal %s at:\n%s" % (
00069                 signum, self.format_frame(frame, limit=1)))
00070         raise HandlerBCalled(signum, self.format_frame(frame))

Here is the caller graph for this function:

Definition at line 81 of file test_signal.py.

00081 
00082     def run_test(self):
00083         # Install handlers. This function runs in a sub-process, so we
00084         # don't worry about re-setting the default handlers.
00085         signal.signal(signal.SIGHUP, self.handlerA)
00086         signal.signal(signal.SIGUSR1, self.handlerB)
00087         signal.signal(signal.SIGUSR2, signal.SIG_IGN)
00088         signal.signal(signal.SIGALRM, signal.default_int_handler)
00089 
00090         # Variables the signals will modify:
00091         self.a_called = False
00092         self.b_called = False
00093 
00094         # Let the sub-processes know who to send signals to.
00095         pid = os.getpid()
00096         if support.verbose:
00097             print("test runner's pid is", pid)
00098 
00099         child = ignoring_eintr(subprocess.Popen, ['kill', '-HUP', str(pid)])
00100         if child:
00101             self.wait(child)
00102             if not self.a_called:
00103                 time.sleep(1)  # Give the signal time to be delivered.
00104         self.assertTrue(self.a_called)
00105         self.assertFalse(self.b_called)
00106         self.a_called = False
00107 
00108         # Make sure the signal isn't delivered while the previous
00109         # Popen object is being destroyed, because __del__ swallows
00110         # exceptions.
00111         del child
00112         try:
00113             child = subprocess.Popen(['kill', '-USR1', str(pid)])
00114             # This wait should be interrupted by the signal's exception.
00115             self.wait(child)
00116             time.sleep(1)  # Give the signal time to be delivered.
00117             self.fail('HandlerBCalled exception not thrown')
00118         except HandlerBCalled:
00119             self.assertTrue(self.b_called)
00120             self.assertFalse(self.a_called)
00121             if support.verbose:
00122                 print("HandlerBCalled exception caught")
00123 
00124         child = ignoring_eintr(subprocess.Popen, ['kill', '-USR2', str(pid)])
00125         if child:
00126             self.wait(child)  # Nothing should happen.
00127 
00128         try:
00129             signal.alarm(1)
00130             # The race condition in pause doesn't matter in this case,
00131             # since alarm is going to raise a KeyboardException, which
00132             # will skip the call.
00133             signal.pause()
00134             # But if another signal arrives before the alarm, pause
00135             # may return early.
00136             time.sleep(1)
00137         except KeyboardInterrupt:
00138             if support.verbose:
00139                 print("KeyboardInterrupt (the alarm() went off)")
00140         except:
00141             self.fail("Some other exception woke us from pause: %s" %
00142                       traceback.format_exc())
00143         else:
00144             self.fail("pause returned of its own accord, and the signal"
00145                       " didn't arrive after another second.")

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 47 of file test_signal.py.

00047 
00048     def setUp(self):
00049         self.using_gc = gc.isenabled()
00050         gc.disable()

Here is the caller graph for this function:

Definition at line 51 of file test_signal.py.

00051 
00052     def tearDown(self):
00053         if self.using_gc:
00054             gc.enable()

Here is the caller graph for this function:

Definition at line 150 of file test_signal.py.

00150 
00151     def test_main(self):
00152         # This function spawns a child process to insulate the main
00153         # test-running process from all the signals. It then
00154         # communicates with that child process over a pipe and
00155         # re-raises information about any exceptions the child
00156         # throws. The real work happens in self.run_test().
00157         os_done_r, os_done_w = os.pipe()
00158         with closing(os.fdopen(os_done_r, 'rb')) as done_r, \
00159              closing(os.fdopen(os_done_w, 'wb')) as done_w:
00160             child = os.fork()
00161             if child == 0:
00162                 # In the child process; run the test and report results
00163                 # through the pipe.
00164                 try:
00165                     done_r.close()
00166                     # Have to close done_w again here because
00167                     # exit_subprocess() will skip the enclosing with block.
00168                     with closing(done_w):
00169                         try:
00170                             self.run_test()
00171                         except:
00172                             pickle.dump(traceback.format_exc(), done_w)
00173                         else:
00174                             pickle.dump(None, done_w)
00175                 except:
00176                     print('Uh oh, raised from pickle.')
00177                     traceback.print_exc()
00178                 finally:
00179                     exit_subprocess()
00180 
00181             done_w.close()
00182             # Block for up to MAX_DURATION seconds for the test to finish.
00183             r, w, x = select.select([done_r], [], [], self.MAX_DURATION)
00184             if done_r in r:
00185                 tb = pickle.load(done_r)
00186                 if tb:
00187                     self.fail(tb)
00188             else:
00189                 os.kill(child, signal.SIGKILL)
00190                 self.fail('Test deadlocked after %d seconds.' %
00191                           self.MAX_DURATION)
00192 
00193 
@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")

Here is the call graph for this function:

def test.test_signal.InterProcessSignalTests.wait (   self,
  child 
)
Wait for child to finish, ignoring EINTR.

Definition at line 71 of file test_signal.py.

00071 
00072     def wait(self, child):
00073         """Wait for child to finish, ignoring EINTR."""
00074         while True:
00075             try:
00076                 child.wait()
00077                 return
00078             except OSError as e:
00079                 if e.errno != errno.EINTR:
00080                     raise

Here is the caller graph for this function:


Member Data Documentation

Definition at line 59 of file test_signal.py.

Definition at line 65 of file test_signal.py.

Definition at line 45 of file test_signal.py.

Definition at line 48 of file test_signal.py.


The documentation for this class was generated from the following file: