0001 """TestCases for using the DB.join and DBCursor.join_item methods. 0002 """ 0003 0004 import sys, os, string 0005 import tempfile 0006 import time 0007 from pprint import pprint 0008 0009 try: 0010 from threading import Thread, currentThread 0011 have_threads = 1 0012 except ImportError: 0013 have_threads = 0 0014 0015 import unittest 0016 from test_all import verbose 0017 0018 try: 0019 # For Pythons w/distutils pybsddb 0020 from bsddb3 import db, dbshelve 0021 except ImportError: 0022 # For Python 2.3 0023 from bsddb import db, dbshelve 0024 0025 0026 #---------------------------------------------------------------------- 0027 0028 ProductIndex = [ 0029 ('apple', "Convenience Store"), 0030 ('blueberry', "Farmer's Market"), 0031 ('shotgun', "S-Mart"), # Aisle 12 0032 ('pear', "Farmer's Market"), 0033 ('chainsaw', "S-Mart"), # "Shop smart. Shop S-Mart!" 0034 ('strawberry', "Farmer's Market"), 0035 ] 0036 0037 ColorIndex = [ 0038 ('blue', "blueberry"), 0039 ('red', "apple"), 0040 ('red', "chainsaw"), 0041 ('red', "strawberry"), 0042 ('yellow', "peach"), 0043 ('yellow', "pear"), 0044 ('black', "shotgun"), 0045 ] 0046 0047 class JoinTestCase(unittest.TestCase): 0048 keytype = '' 0049 0050 def setUp(self): 0051 self.filename = self.__class__.__name__ + '.db' 0052 homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home') 0053 self.homeDir = homeDir 0054 try: os.mkdir(homeDir) 0055 except os.error: pass 0056 self.env = db.DBEnv() 0057 self.env.open(homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOCK ) 0058 0059 def tearDown(self): 0060 self.env.close() 0061 import glob 0062 files = glob.glob(os.path.join(self.homeDir, '*')) 0063 for file in files: 0064 os.remove(file) 0065 0066 def test01_join(self): 0067 if verbose: 0068 print '\n', '-=' * 30 0069 print "Running %s.test01_join..." % \ 0070 self.__class__.__name__ 0071 0072 # create and populate primary index 0073 priDB = db.DB(self.env) 0074 priDB.open(self.filename, "primary", db.DB_BTREE, db.DB_CREATE) 0075 map(lambda t, priDB=priDB: apply(priDB.put, t), ProductIndex) 0076 0077 # create and populate secondary index 0078 secDB = db.DB(self.env) 0079 secDB.set_flags(db.DB_DUP | db.DB_DUPSORT) 0080 secDB.open(self.filename, "secondary", db.DB_BTREE, db.DB_CREATE) 0081 map(lambda t, secDB=secDB: apply(secDB.put, t), ColorIndex) 0082 0083 sCursor = None 0084 jCursor = None 0085 try: 0086 # lets look up all of the red Products 0087 sCursor = secDB.cursor() 0088 # Don't do the .set() in an assert, or you can get a bogus failure 0089 # when running python -O 0090 tmp = sCursor.set('red') 0091 assert tmp 0092 0093 # FIXME: jCursor doesn't properly hold a reference to its 0094 # cursors, if they are closed before jcursor is used it 0095 # can cause a crash. 0096 jCursor = priDB.join([sCursor]) 0097 0098 if jCursor.get(0) != ('apple', "Convenience Store"): 0099 self.fail("join cursor positioned wrong") 0100 if jCursor.join_item() != 'chainsaw': 0101 self.fail("DBCursor.join_item returned wrong item") 0102 if jCursor.get(0)[0] != 'strawberry': 0103 self.fail("join cursor returned wrong thing") 0104 if jCursor.get(0): # there were only three red items to return 0105 self.fail("join cursor returned too many items") 0106 finally: 0107 if jCursor: 0108 jCursor.close() 0109 if sCursor: 0110 sCursor.close() 0111 priDB.close() 0112 secDB.close() 0113 0114 0115 def test_suite(): 0116 suite = unittest.TestSuite() 0117 0118 suite.addTest(unittest.makeSuite(JoinTestCase)) 0119 0120 return suite 0121
Generated by PyXR 0.9.4