]> git.decadent.org.uk Git - dak.git/blob - tests/dbtest_multiproc.py
Merge remote branch 'mhy/master' into merge
[dak.git] / tests / dbtest_multiproc.py
1 #!/usr/bin/env python
2
3 from db_test import DBDakTestCase
4
5 from daklib.dbconn import DBConn
6
7 from multiprocessing import Pool
8 from time import sleep
9 import unittest
10
11 def read_number():
12     DBConn().reset()
13     session = DBConn().session()
14     result = session.query('foo').from_statement('select 7 as foo').scalar()
15     sleep(0.1)
16     session.close()
17     return result
18
19 class MultiProcTestCase(DBDakTestCase):
20     """
21     This TestCase checks that DBConn works with multiprocessing. A fresh
22     subprocess needs to call reset() on DBConn(). See function read_number()
23     for an example.
24     """
25
26     def save_result(self, result):
27         self.result += result
28
29     def test_seven(self):
30         '''
31         Test apply_async() with a database session.
32         '''
33         self.result = 0
34         pool = Pool()
35         pool.apply_async(read_number, (), callback = self.save_result)
36         pool.apply_async(read_number, (), callback = self.save_result)
37         pool.apply_async(read_number, (), callback = self.save_result)
38         pool.apply_async(read_number, (), callback = self.save_result)
39         pool.apply_async(read_number, (), callback = self.save_result)
40         pool.close()
41         pool.join()
42         self.assertEqual(5 * 7, self.result)
43
44 if __name__ == '__main__':
45     unittest.main()