Home | History | Annotate | Download | only in test 
      1 """TestCases for using the DB.join and DBCursor.join_item methods.
      2 """
      3 
      4 import os
      5 
      6 import unittest
      7 
      8 from test_all import db, dbshelve, test_support, verbose, \
      9         get_new_environment_path, get_new_database_path
     10 
     11 #----------------------------------------------------------------------
     12 
     13 ProductIndex = [
     14     ('apple', "Convenience Store"),
     15     ('blueberry', "Farmer's Market"),
     16     ('shotgun', "S-Mart"),              # Aisle 12
     17     ('pear', "Farmer's Market"),
     18     ('chainsaw', "S-Mart"),             # "Shop smart.  Shop S-Mart!"
     19     ('strawberry', "Farmer's Market"),
     20 ]
     21 
     22 ColorIndex = [
     23     ('blue', "blueberry"),
     24     ('red', "apple"),
     25     ('red', "chainsaw"),
     26     ('red', "strawberry"),
     27     ('yellow', "peach"),
     28     ('yellow', "pear"),
     29     ('black', "shotgun"),
     30 ]
     31 
     32 class JoinTestCase(unittest.TestCase):
     33     keytype = ''
     34 
     35     def setUp(self):
     36         self.filename = self.__class__.__name__ + '.db'
     37         self.homeDir = get_new_environment_path()
     38         self.env = db.DBEnv()
     39         self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOCK )
     40 
     41     def tearDown(self):
     42         self.env.close()
     43         test_support.rmtree(self.homeDir)
     44 
     45     def test01_join(self):
     46         if verbose:
     47             print '\n', '-=' * 30
     48             print "Running %s.test01_join..." % \
     49                   self.__class__.__name__
     50 
     51         # create and populate primary index
     52         priDB = db.DB(self.env)
     53         priDB.open(self.filename, "primary", db.DB_BTREE, db.DB_CREATE)
     54         map(lambda t, priDB=priDB: priDB.put(*t), ProductIndex)
     55 
     56         # create and populate secondary index
     57         secDB = db.DB(self.env)
     58         secDB.set_flags(db.DB_DUP | db.DB_DUPSORT)
     59         secDB.open(self.filename, "secondary", db.DB_BTREE, db.DB_CREATE)
     60         map(lambda t, secDB=secDB: secDB.put(*t), ColorIndex)
     61 
     62         sCursor = None
     63         jCursor = None
     64         try:
     65             # lets look up all of the red Products
     66             sCursor = secDB.cursor()
     67             # Don't do the .set() in an assert, or you can get a bogus failure
     68             # when running python -O
     69             tmp = sCursor.set('red')
     70             self.assertTrue(tmp)
     71 
     72             # FIXME: jCursor doesn't properly hold a reference to its
     73             # cursors, if they are closed before jcursor is used it
     74             # can cause a crash.
     75             jCursor = priDB.join([sCursor])
     76 
     77             if jCursor.get(0) != ('apple', "Convenience Store"):
     78                 self.fail("join cursor positioned wrong")
     79             if jCursor.join_item() != 'chainsaw':
     80                 self.fail("DBCursor.join_item returned wrong item")
     81             if jCursor.get(0)[0] != 'strawberry':
     82                 self.fail("join cursor returned wrong thing")
     83             if jCursor.get(0):  # there were only three red items to return
     84                 self.fail("join cursor returned too many items")
     85         finally:
     86             if jCursor:
     87                 jCursor.close()
     88             if sCursor:
     89                 sCursor.close()
     90             priDB.close()
     91             secDB.close()
     92 
     93 
     94 def test_suite():
     95     suite = unittest.TestSuite()
     96 
     97     suite.addTest(unittest.makeSuite(JoinTestCase))
     98 
     99     return suite
    100