]> git.decadent.org.uk Git - dak.git/blob - tests/test_multiprocessing.py
Merge remote branch 'drkranz/show-new' into merge
[dak.git] / 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])