]> git.decadent.org.uk Git - dak.git/blob - tests/dbtest_multiproc.py
Refactor functions getSource() and getBinaries().
[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     session = DBConn().session()
13     result = session.query('foo').from_statement('select 7 as foo').scalar()
14     sleep(0.1)
15     session.close()
16     return result
17
18 class MultiProcTestCase(DBDakTestCase):
19     """
20     This TestCase checks that DBConn works with multiprocessing.
21     """
22
23     def save_result(self, result):
24         self.result += result
25
26     def test_seven(self):
27         '''
28         Test apply_async() with a database session.
29         '''
30         self.result = 0
31         pool = Pool()
32         pool.apply_async(read_number, (), callback = self.save_result)
33         pool.apply_async(read_number, (), callback = self.save_result)
34         pool.apply_async(read_number, (), callback = self.save_result)
35         pool.apply_async(read_number, (), callback = self.save_result)
36         pool.apply_async(read_number, (), callback = self.save_result)
37         pool.close()
38         pool.join()
39         self.assertEqual(5 * 7, self.result)
40
41 if __name__ == '__main__':
42     unittest.main()