Initial commit of OpenSPARC T2 architecture model.

[OpenSPARC-T2-SAM] / sam-t2 / devtools / amd64 / lib / python2.4 / bsddb / test / test_lock.py

"""
TestCases for testing the locking sub-system.
"""

import sys, os, string
import tempfile
import time
from pprint import pprint

try:
   from threading import Thread, currentThread
   have_threads = 1
except ImportError:
   have_threads = 0


import unittest
from test_all import verbose

try:
   # For Pythons w/distutils pybsddb
   from bsddb3 import db
except ImportError:
   # For Python 2.3
   from bsddb import db


#----------------------------------------------------------------------

class LockingTestCase(unittest.TestCase):

   def setUp(self):
       homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home')
       self.homeDir = homeDir
       try: os.mkdir(homeDir)
       except os.error: pass
       self.env = db.DBEnv()
       self.env.open(homeDir, db.DB_THREAD | db.DB_INIT_MPOOL |
                     db.DB_INIT_LOCK | db.DB_CREATE)


   def tearDown(self):
       self.env.close()
       import glob
       files = glob.glob(os.path.join(self.homeDir, '*'))
       for file in files:
           os.remove(file)


   def test01_simple(self):
       if verbose:
           print '\n', '-=' * 30
           print "Running %s.test01_simple..." % self.__class__.__name__

       anID = self.env.lock_id()
       if verbose:
           print "locker ID: %s" % anID
       lock = self.env.lock_get(anID, "some locked thing", db.DB_LOCK_WRITE)
       if verbose:
           print "Aquired lock: %s" % lock
       time.sleep(1)
       self.env.lock_put(lock)
       if verbose:
           print "Released lock: %s" % lock




   def test02_threaded(self):
       if verbose:
           print '\n', '-=' * 30
           print "Running %s.test02_threaded..." % self.__class__.__name__

       threads = []
       threads.append(Thread(target = self.theThread,
                             args=(5, db.DB_LOCK_WRITE)))
       threads.append(Thread(target = self.theThread,
                             args=(1, db.DB_LOCK_READ)))
       threads.append(Thread(target = self.theThread,
                             args=(1, db.DB_LOCK_READ)))
       threads.append(Thread(target = self.theThread,
                             args=(1, db.DB_LOCK_WRITE)))
       threads.append(Thread(target = self.theThread,
                             args=(1, db.DB_LOCK_READ)))
       threads.append(Thread(target = self.theThread,
                             args=(1, db.DB_LOCK_READ)))
       threads.append(Thread(target = self.theThread,
                             args=(1, db.DB_LOCK_WRITE)))
       threads.append(Thread(target = self.theThread,
                             args=(1, db.DB_LOCK_WRITE)))
       threads.append(Thread(target = self.theThread,
                             args=(1, db.DB_LOCK_WRITE)))

       for t in threads:
           t.start()
       for t in threads:
           t.join()

   def test03_set_timeout(self):
       # test that the set_timeout call works
       if hasattr(self.env, 'set_timeout'):
           self.env.set_timeout(0, db.DB_SET_LOCK_TIMEOUT)
           self.env.set_timeout(0, db.DB_SET_TXN_TIMEOUT)
           self.env.set_timeout(123456, db.DB_SET_LOCK_TIMEOUT)
           self.env.set_timeout(7890123, db.DB_SET_TXN_TIMEOUT)

   def theThread(self, sleepTime, lockType):
       name = currentThread().getName()
       if lockType ==  db.DB_LOCK_WRITE:
           lt = "write"
       else:
           lt = "read"

       anID = self.env.lock_id()
       if verbose:
           print "%s: locker ID: %s" % (name, anID)

       lock = self.env.lock_get(anID, "some locked thing", lockType)
       if verbose:
           print "%s: Aquired %s lock: %s" % (name, lt, lock)

       time.sleep(sleepTime)

       self.env.lock_put(lock)
       if verbose:
           print "%s: Released %s lock: %s" % (name, lt, lock)


#----------------------------------------------------------------------

def test_suite():
   suite = unittest.TestSuite()

   if have_threads:
       suite.addTest(unittest.makeSuite(LockingTestCase))
   else:
       suite.addTest(unittest.makeSuite(LockingTestCase, 'test01'))

   return suite


if __name__ == '__main__':
   unittest.main(defaultTest='test_suite')