tests/test_multiprocessing.py

   1 #!/usr/bin/python
   2
   3 from base_test import DakTestCase
   4
   5 from daklib.dakmultiprocessing import DakProcessPool, \
   6                                       PROC_STATUS_SUCCESS,   PROC_STATUS_MISCFAILURE, \
   7                                       PROC_STATUS_EXCEPTION, PROC_STATUS_SIGNALRAISED
   8 import signal
   9
  10 def test_function(num, num2):
  11     from os import kill, getpid
  12
  13     if num == 1:
  14         sigs = [signal.SIGTERM, signal.SIGPIPE, signal.SIGALRM, signal.SIGHUP]
  15         kill(getpid(), sigs[num2])
  16
  17     if num2 == 3:
  18         raise Exception('Test uncaught exception handling')
  19
  20     if num == 0 and num2 == 1:
  21         return (PROC_STATUS_MISCFAILURE, 'Test custom error return')
  22
  23     return (PROC_STATUS_SUCCESS, 'blah, %d, %d' % (num, num2))
  24
  25 class DakProcessPoolTestCase(DakTestCase):
  26     def testPool(self):
  27         def alarm_handler(signum, frame):
  28             raise AssertionError('Timed out')
  29
  30         # Shouldn't take us more than 15 seconds to run this test
  31         signal.signal(signal.SIGALRM, alarm_handler)
  32         signal.alarm(15)
  33
  34         p = DakProcessPool()
  35         for s in range(3):
  36             for j in range(4):
  37                 p.apply_async(test_function, [s, j])
  38
  39         p.close()
  40         p.join()
  41
  42         signal.alarm(0)
  43         signal.signal(signal.SIGALRM, signal.SIG_DFL)
  44
  45         expected = [(PROC_STATUS_SUCCESS,      'blah, 0, 0'),
  46                     (PROC_STATUS_MISCFAILURE,  'Test custom error return'),
  47                     (PROC_STATUS_SUCCESS,      'blah, 0, 2'),
  48                     (PROC_STATUS_EXCEPTION,    'Test uncaught exception handling'),
  49                     (PROC_STATUS_SIGNALRAISED, 15),
  50                     (PROC_STATUS_SIGNALRAISED, 13),
  51                     (PROC_STATUS_SIGNALRAISED, 14),
  52                     (PROC_STATUS_SIGNALRAISED, 1),
  53                     (PROC_STATUS_SUCCESS,      'blah, 2, 0'),
  54                     (PROC_STATUS_SUCCESS,      'blah, 2, 1'),
  55                     (PROC_STATUS_SUCCESS,      'blah, 2, 2'),
  56                     (PROC_STATUS_EXCEPTION,    'Test uncaught exception handling')]
  57
  58         self.assertEqual( len(p.results), len(expected) )
  59
  60         for r in range(len(p.results)):
  61             self.assertEqual(p.results[r], expected[r])