cpython/Lib/bsddb/test/test_join.py
Neal Norwitz 6057b2e645 Create a db_home directory with a unique name so multiple users can
run the test simultaneously.  The simplest thing I found that worked
on both Windows and Unix was to use the PID.  It's unique so should be
sufficient.  This should prevent many of the spurious failures of
the automated tests since they run as different users.

Also cleanup the directory consistenly in the tearDown methods.

It would be nice if someone ensured that the directories are always
created with a consistent name.
2008-02-24 18:47:03 +00:00

116 lines
3.4 KiB
Python

"""TestCases for using the DB.join and DBCursor.join_item methods.
"""
import os
import tempfile
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, dbshelve
except ImportError:
# For Python 2.3
from bsddb import db, dbshelve
#----------------------------------------------------------------------
ProductIndex = [
('apple', "Convenience Store"),
('blueberry', "Farmer's Market"),
('shotgun', "S-Mart"), # Aisle 12
('pear', "Farmer's Market"),
('chainsaw', "S-Mart"), # "Shop smart. Shop S-Mart!"
('strawberry', "Farmer's Market"),
]
ColorIndex = [
('blue', "blueberry"),
('red', "apple"),
('red', "chainsaw"),
('red', "strawberry"),
('yellow', "peach"),
('yellow', "pear"),
('black', "shotgun"),
]
class JoinTestCase(unittest.TestCase):
keytype = ''
def setUp(self):
self.filename = self.__class__.__name__ + '.db'
homeDir = os.path.join(tempfile.gettempdir(), 'db_home%d'%os.getpid())
self.homeDir = homeDir
try: os.mkdir(homeDir)
except os.error: pass
self.env = db.DBEnv()
self.env.open(homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOCK )
def tearDown(self):
self.env.close()
from test import test_support
test_support.rmtree(self.homeDir)
def test01_join(self):
if verbose:
print '\n', '-=' * 30
print "Running %s.test01_join..." % \
self.__class__.__name__
# create and populate primary index
priDB = db.DB(self.env)
priDB.open(self.filename, "primary", db.DB_BTREE, db.DB_CREATE)
map(lambda t, priDB=priDB: apply(priDB.put, t), ProductIndex)
# create and populate secondary index
secDB = db.DB(self.env)
secDB.set_flags(db.DB_DUP | db.DB_DUPSORT)
secDB.open(self.filename, "secondary", db.DB_BTREE, db.DB_CREATE)
map(lambda t, secDB=secDB: apply(secDB.put, t), ColorIndex)
sCursor = None
jCursor = None
try:
# lets look up all of the red Products
sCursor = secDB.cursor()
# Don't do the .set() in an assert, or you can get a bogus failure
# when running python -O
tmp = sCursor.set('red')
assert tmp
# FIXME: jCursor doesn't properly hold a reference to its
# cursors, if they are closed before jcursor is used it
# can cause a crash.
jCursor = priDB.join([sCursor])
if jCursor.get(0) != ('apple', "Convenience Store"):
self.fail("join cursor positioned wrong")
if jCursor.join_item() != 'chainsaw':
self.fail("DBCursor.join_item returned wrong item")
if jCursor.get(0)[0] != 'strawberry':
self.fail("join cursor returned wrong thing")
if jCursor.get(0): # there were only three red items to return
self.fail("join cursor returned too many items")
finally:
if jCursor:
jCursor.close()
if sCursor:
sCursor.close()
priDB.close()
secDB.close()
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(JoinTestCase))
return suite