"""TestCases for using the DB.join and DBCursor.join_item methods.
from pprint
import pprint
from threading
import Thread
, currentThread
from test_all
import verbose
# For Pythons w/distutils pybsddb
from bsddb3
import db
, dbshelve
from bsddb
import db
, dbshelve
#----------------------------------------------------------------------
('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"),
class JoinTestCase(unittest
.TestCase
):
self
.filename
= self
.__class
__.__name
__ + '.db'
homeDir
= os
.path
.join(os
.path
.dirname(sys
.argv
[0]), 'db_home')
self
.env
.open(homeDir
, db
.DB_CREATE | db
.DB_INIT_MPOOL | db
.DB_INIT_LOCK
)
files
= glob
.glob(os
.path
.join(self
.homeDir
, '*'))
print "Running %s.test01_join..." % \
# create and populate primary index
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
.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
)
# lets look up all of the red Products
# Don't do the .set() in an assert, or you can get a bogus failure
# FIXME: jCursor doesn't properly hold a reference to its
# cursors, if they are closed before jcursor is used it
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")
suite
= unittest
.TestSuite()
suite
.addTest(unittest
.makeSuite(JoinTestCase
))