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