Lib/bsddb/test/test_join.py

   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: apply(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: apply(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.assert_(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