TestCases for testing the locking sub-system.
from pprint
import pprint
from threading
import Thread
, currentThread
from test_all
import verbose
# For Pythons w/distutils pybsddb
#----------------------------------------------------------------------
class LockingTestCase(unittest
.TestCase
):
homeDir
= os
.path
.join(os
.path
.dirname(sys
.argv
[0]), 'db_home')
self
.env
.open(homeDir
, db
.DB_THREAD | db
.DB_INIT_MPOOL |
db
.DB_INIT_LOCK | db
.DB_CREATE
)
files
= glob
.glob(os
.path
.join(self
.homeDir
, '*'))
print "Running %s.test01_simple..." % self
.__class
__.__name
__
anID
= self
.env
.lock_id()
print "locker ID: %s" % anID
lock
= self
.env
.lock_get(anID
, "some locked thing", db
.DB_LOCK_WRITE
)
print "Aquired lock: %s" % lock
print "Released lock: %s" % lock
def test02_threaded(self
):
print "Running %s.test02_threaded..." % self
.__class
__.__name
__
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
)))
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
:
anID
= self
.env
.lock_id()
print "%s: locker ID: %s" % (name
, anID
)
lock
= self
.env
.lock_get(anID
, "some locked thing", lockType
)
print "%s: Aquired %s lock: %s" % (name
, lt
, lock
)
print "%s: Released %s lock: %s" % (name
, lt
, lock
)
#----------------------------------------------------------------------
suite
= unittest
.TestSuite()
suite
.addTest(unittest
.makeSuite(LockingTestCase
))
suite
.addTest(unittest
.makeSuite(LockingTestCase
, 'test01'))
if __name__
== '__main__':
unittest
.main(defaultTest
='test_suite')