diff -r ffa851df0825 -r 2fb8b9db1c86 symbian-qemu-0.9.1-12/python-2.6.1/Lib/bsddb/test/test_join.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/symbian-qemu-0.9.1-12/python-2.6.1/Lib/bsddb/test/test_join.py Fri Jul 31 15:01:17 2009 +0100 @@ -0,0 +1,99 @@ +"""TestCases for using the DB.join and DBCursor.join_item methods. +""" + +import os + +import unittest + +from test_all import db, dbshelve, test_support, verbose, \ + get_new_environment_path, get_new_database_path + +#---------------------------------------------------------------------- + +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' + self.homeDir = get_new_environment_path() + self.env = db.DBEnv() + self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOCK ) + + def tearDown(self): + self.env.close() + 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') + self.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