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