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()