home *** CD-ROM | disk | FTP | other *** search
/ PC World 2005 June / PCWorld_2005-06_cd.bin / software / vyzkuste / firewally / firewally.exe / framework-2.3.exe / test_lock.py < prev    next >
Text File  |  2003-12-30  |  4KB  |  145 lines
  1. """
  2. TestCases for testing the locking sub-system.
  3. """
  4.  
  5. import sys, os, string
  6. import tempfile
  7. import time
  8. from pprint import pprint
  9. from whrandom import random
  10.  
  11. try:
  12.     from threading import Thread, currentThread
  13.     have_threads = 1
  14. except ImportError:
  15.     have_threads = 0
  16.  
  17.  
  18. import unittest
  19. from test_all import verbose
  20.  
  21. try:
  22.     # For Python 2.3
  23.     from bsddb import db
  24. except ImportError:
  25.     # For earlier Pythons w/distutils pybsddb
  26.     from bsddb3 import db
  27.  
  28.  
  29. #----------------------------------------------------------------------
  30.  
  31. class LockingTestCase(unittest.TestCase):
  32.  
  33.     def setUp(self):
  34.         homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home')
  35.         self.homeDir = homeDir
  36.         try: os.mkdir(homeDir)
  37.         except os.error: pass
  38.         self.env = db.DBEnv()
  39.         self.env.open(homeDir, db.DB_THREAD | db.DB_INIT_MPOOL |
  40.                       db.DB_INIT_LOCK | db.DB_CREATE)
  41.  
  42.  
  43.     def tearDown(self):
  44.         self.env.close()
  45.         import glob
  46.         files = glob.glob(os.path.join(self.homeDir, '*'))
  47.         for file in files:
  48.             os.remove(file)
  49.  
  50.  
  51.     def test01_simple(self):
  52.         if verbose:
  53.             print '\n', '-=' * 30
  54.             print "Running %s.test01_simple..." % self.__class__.__name__
  55.  
  56.         anID = self.env.lock_id()
  57.         if verbose:
  58.             print "locker ID: %s" % anID
  59.         lock = self.env.lock_get(anID, "some locked thing", db.DB_LOCK_WRITE)
  60.         if verbose:
  61.             print "Aquired lock: %s" % lock
  62.         time.sleep(1)
  63.         self.env.lock_put(lock)
  64.         if verbose:
  65.             print "Released lock: %s" % lock
  66.  
  67.  
  68.  
  69.  
  70.     def test02_threaded(self):
  71.         if verbose:
  72.             print '\n', '-=' * 30
  73.             print "Running %s.test02_threaded..." % self.__class__.__name__
  74.  
  75.         threads = []
  76.         threads.append(Thread(target = self.theThread,
  77.                               args=(5, db.DB_LOCK_WRITE)))
  78.         threads.append(Thread(target = self.theThread,
  79.                               args=(1, db.DB_LOCK_READ)))
  80.         threads.append(Thread(target = self.theThread,
  81.                               args=(1, db.DB_LOCK_READ)))
  82.         threads.append(Thread(target = self.theThread,
  83.                               args=(1, db.DB_LOCK_WRITE)))
  84.         threads.append(Thread(target = self.theThread,
  85.                               args=(1, db.DB_LOCK_READ)))
  86.         threads.append(Thread(target = self.theThread,
  87.                               args=(1, db.DB_LOCK_READ)))
  88.         threads.append(Thread(target = self.theThread,
  89.                               args=(1, db.DB_LOCK_WRITE)))
  90.         threads.append(Thread(target = self.theThread,
  91.                               args=(1, db.DB_LOCK_WRITE)))
  92.         threads.append(Thread(target = self.theThread,
  93.                               args=(1, db.DB_LOCK_WRITE)))
  94.  
  95.         for t in threads:
  96.             t.start()
  97.         for t in threads:
  98.             t.join()
  99.  
  100.     def test03_set_timeout(self):
  101.         # test that the set_timeout call works
  102.         if hasattr(self.env, 'set_timeout'):
  103.             self.env.set_timeout(0, db.DB_SET_LOCK_TIMEOUT)
  104.             self.env.set_timeout(0, db.DB_SET_TXN_TIMEOUT)
  105.             self.env.set_timeout(123456, db.DB_SET_LOCK_TIMEOUT)
  106.             self.env.set_timeout(7890123, db.DB_SET_TXN_TIMEOUT)
  107.  
  108.     def theThread(self, sleepTime, lockType):
  109.         name = currentThread().getName()
  110.         if lockType ==  db.DB_LOCK_WRITE:
  111.             lt = "write"
  112.         else:
  113.             lt = "read"
  114.  
  115.         anID = self.env.lock_id()
  116.         if verbose:
  117.             print "%s: locker ID: %s" % (name, anID)
  118.  
  119.         lock = self.env.lock_get(anID, "some locked thing", lockType)
  120.         if verbose:
  121.             print "%s: Aquired %s lock: %s" % (name, lt, lock)
  122.  
  123.         time.sleep(sleepTime)
  124.  
  125.         self.env.lock_put(lock)
  126.         if verbose:
  127.             print "%s: Released %s lock: %s" % (name, lt, lock)
  128.  
  129.  
  130. #----------------------------------------------------------------------
  131.  
  132. def test_suite():
  133.     suite = unittest.TestSuite()
  134.  
  135.     if have_threads:
  136.         suite.addTest(unittest.makeSuite(LockingTestCase))
  137.     else:
  138.         suite.addTest(unittest.makeSuite(LockingTestCase, 'test01'))
  139.  
  140.     return suite
  141.  
  142.  
  143. if __name__ == '__main__':
  144.     unittest.main(defaultTest='test_suite')
  145.