""" TestCases for DB.associate. """ 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, dbshelve except ImportError: # For Python 2.3 from bsddb import db, dbshelve #---------------------------------------------------------------------- musicdata = { 1 : ("Bad English", "The Price Of Love", "Rock"), 2 : ("DNA featuring Suzanne Vega", "Tom's Diner", "Rock"), 3 : ("George Michael", "Praying For Time", "Rock"), 4 : ("Gloria Estefan", "Here We Are", "Rock"), 5 : ("Linda Ronstadt", "Don't Know Much", "Rock"), 6 : ("Michael Bolton", "How Am I Supposed To Live Without You", "Blues"), 7 : ("Paul Young", "Oh Girl", "Rock"), 8 : ("Paula Abdul", "Opposites Attract", "Rock"), 9 : ("Richard Marx", "Should've Known Better", "Rock"), 10: ("Rod Stewart", "Forever Young", "Rock"), 11: ("Roxette", "Dangerous", "Rock"), 12: ("Sheena Easton", "The Lover In Me", "Rock"), 13: ("Sinead O'Connor", "Nothing Compares 2 U", "Rock"), 14: ("Stevie B.", "Because I Love You", "Rock"), 15: ("Taylor Dayne", "Love Will Lead You Back", "Rock"), 16: ("The Bangles", "Eternal Flame", "Rock"), 17: ("Wilson Phillips", "Release Me", "Rock"), 18: ("Billy Joel", "Blonde Over Blue", "Rock"), 19: ("Billy Joel", "Famous Last Words", "Rock"), 20: ("Billy Joel", "Lullabye (Goodnight, My Angel)", "Rock"), 21: ("Billy Joel", "The River Of Dreams", "Rock"), 22: ("Billy Joel", "Two Thousand Years", "Rock"), 23: ("Janet Jackson", "Alright", "Rock"), 24: ("Janet Jackson", "Black Cat", "Rock"), 25: ("Janet Jackson", "Come Back To Me", "Rock"), 26: ("Janet Jackson", "Escapade", "Rock"), 27: ("Janet Jackson", "Love Will Never Do (Without You)", "Rock"), 28: ("Janet Jackson", "Miss You Much", "Rock"), 29: ("Janet Jackson", "Rhythm Nation", "Rock"), 30: ("Janet Jackson", "State Of The World", "Rock"), 31: ("Janet Jackson", "The Knowledge", "Rock"), 32: ("Spyro Gyra", "End of Romanticism", "Jazz"), 33: ("Spyro Gyra", "Heliopolis", "Jazz"), 34: ("Spyro Gyra", "Jubilee", "Jazz"), 35: ("Spyro Gyra", "Little Linda", "Jazz"), 36: ("Spyro Gyra", "Morning Dance", "Jazz"), 37: ("Spyro Gyra", "Song for Lorraine", "Jazz"), 38: ("Yes", "Owner Of A Lonely Heart", "Rock"), 39: ("Yes", "Rhythm Of Love", "Rock"), 40: ("Cusco", "Dream Catcher", "New Age"), 41: ("Cusco", "Geronimos Laughter", "New Age"), 42: ("Cusco", "Ghost Dance", "New Age"), 43: ("Blue Man Group", "Drumbone", "New Age"), 44: ("Blue Man Group", "Endless Column", "New Age"), 45: ("Blue Man Group", "Klein Mandelbrot", "New Age"), 46: ("Kenny G", "Silhouette", "Jazz"), 47: ("Sade", "Smooth Operator", "Jazz"), 48: ("David Arkenstone", "Papillon (On The Wings Of The Butterfly)", "New Age"), 49: ("David Arkenstone", "Stepping Stars", "New Age"), 50: ("David Arkenstone", "Carnation Lily Lily Rose", "New Age"), 51: ("David Lanz", "Behind The Waterfall", "New Age"), 52: ("David Lanz", "Cristofori's Dream", "New Age"), 53: ("David Lanz", "Heartsounds", "New Age"), 54: ("David Lanz", "Leaves on the Seine", "New Age"), } #---------------------------------------------------------------------- class AssociateTestCase(unittest.TestCase): keytype = '' def setUp(self): self.filename = self.__class__.__name__ + '.db' 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_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOCK | db.DB_THREAD) def tearDown(self): self.closeDB() self.env.close() import glob files = glob.glob(os.path.join(self.homeDir, '*')) for file in files: os.remove(file) def addDataToDB(self, d): for key, value in musicdata.items(): if type(self.keytype) == type(''): key = "%02d" % key d.put(key, string.join(value, '|')) def createDB(self): self.primary = db.DB(self.env) self.primary.open(self.filename, "primary", self.dbtype, db.DB_CREATE | db.DB_THREAD) def closeDB(self): self.primary.close() def getDB(self): return self.primary def test01_associateWithDB(self): if verbose: print '\n', '-=' * 30 print "Running %s.test01_associateWithDB..." % \ self.__class__.__name__ self.createDB() secDB = db.DB(self.env) secDB.set_flags(db.DB_DUP) secDB.open(self.filename, "secondary", db.DB_BTREE, db.DB_CREATE | db.DB_THREAD) self.getDB().associate(secDB, self.getGenre) self.addDataToDB(self.getDB()) self.finish_test(secDB) def test02_associateAfterDB(self): if verbose: print '\n', '-=' * 30 print "Running %s.test02_associateAfterDB..." % \ self.__class__.__name__ self.createDB() self.addDataToDB(self.getDB()) secDB = db.DB(self.env) secDB.set_flags(db.DB_DUP) secDB.open(self.filename, "secondary", db.DB_BTREE, db.DB_CREATE | db.DB_THREAD) # adding the DB_CREATE flag will cause it to index existing records self.getDB().associate(secDB, self.getGenre, db.DB_CREATE) self.finish_test(secDB) def finish_test(self, secDB): if verbose: print "Primary key traversal:" c = self.getDB().cursor() count = 0 rec = c.first() while rec is not None: if type(self.keytype) == type(''): assert string.atoi(rec[0]) # for primary db, key is a number else: assert rec[0] and type(rec[0]) == type(0) count = count + 1 if verbose: print rec rec = c.next() assert count == len(musicdata) # all items accounted for if verbose: print "Secondary key traversal:" c = secDB.cursor() count = 0 rec = c.first() assert rec[0] == "Jazz" while rec is not None: count = count + 1 if verbose: print rec rec = c.next() # all items accounted for EXCEPT for 1 with "Blues" genre assert count == len(musicdata)-1 def getGenre(self, priKey, priData): assert type(priData) == type("") if verbose: print 'getGenre key:', `priKey`, 'data:', `priData` genre = string.split(priData, '|')[2] if genre == 'Blues': return db.DB_DONOTINDEX else: return genre #---------------------------------------------------------------------- class AssociateHashTestCase(AssociateTestCase): dbtype = db.DB_HASH class AssociateBTreeTestCase(AssociateTestCase): dbtype = db.DB_BTREE class AssociateRecnoTestCase(AssociateTestCase): dbtype = db.DB_RECNO keytype = 0 #---------------------------------------------------------------------- class ShelveAssociateTestCase(AssociateTestCase): def createDB(self): self.primary = dbshelve.open(self.filename, dbname="primary", dbenv=self.env, filetype=self.dbtype) def addDataToDB(self, d): for key, value in musicdata.items(): if type(self.keytype) == type(''): key = "%02d" % key d.put(key, value) # save the value as is this time def getGenre(self, priKey, priData): assert type(priData) == type(()) if verbose: print 'getGenre key:', `priKey`, 'data:', `priData` genre = priData[2] if genre == 'Blues': return db.DB_DONOTINDEX else: return genre class ShelveAssociateHashTestCase(ShelveAssociateTestCase): dbtype = db.DB_HASH class ShelveAssociateBTreeTestCase(ShelveAssociateTestCase): dbtype = db.DB_BTREE class ShelveAssociateRecnoTestCase(ShelveAssociateTestCase): dbtype = db.DB_RECNO keytype = 0 #---------------------------------------------------------------------- class ThreadedAssociateTestCase(AssociateTestCase): def addDataToDB(self, d): t1 = Thread(target = self.writer1, args = (d, )) t2 = Thread(target = self.writer2, args = (d, )) t1.start() t2.start() t1.join() t2.join() def writer1(self, d): for key, value in musicdata.items(): if type(self.keytype) == type(''): key = "%02d" % key d.put(key, string.join(value, '|')) def writer2(self, d): for x in range(100, 600): key = 'z%2d' % x value = [key] * 4 d.put(key, string.join(value, '|')) class ThreadedAssociateHashTestCase(ShelveAssociateTestCase): dbtype = db.DB_HASH class ThreadedAssociateBTreeTestCase(ShelveAssociateTestCase): dbtype = db.DB_BTREE class ThreadedAssociateRecnoTestCase(ShelveAssociateTestCase): dbtype = db.DB_RECNO keytype = 0 #---------------------------------------------------------------------- def test_suite(): suite = unittest.TestSuite() if db.version() >= (3, 3, 11): suite.addTest(unittest.makeSuite(AssociateHashTestCase)) suite.addTest(unittest.makeSuite(AssociateBTreeTestCase)) suite.addTest(unittest.makeSuite(AssociateRecnoTestCase)) suite.addTest(unittest.makeSuite(ShelveAssociateHashTestCase)) suite.addTest(unittest.makeSuite(ShelveAssociateBTreeTestCase)) suite.addTest(unittest.makeSuite(ShelveAssociateRecnoTestCase)) if have_threads: suite.addTest(unittest.makeSuite(ThreadedAssociateHashTestCase)) suite.addTest(unittest.makeSuite(ThreadedAssociateBTreeTestCase)) suite.addTest(unittest.makeSuite(ThreadedAssociateRecnoTestCase)) return suite if __name__ == '__main__': unittest.main(defaultTest='test_suite')