• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""TestCases for using the DB.join and DBCursor.join_item methods.
2"""
3
4import os
5
6import unittest
7
8from test_all import db, dbshelve, test_support, verbose, \
9        get_new_environment_path, get_new_database_path
10
11#----------------------------------------------------------------------
12
13ProductIndex = [
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
22ColorIndex = [
23    ('blue', "blueberry"),
24    ('red', "apple"),
25    ('red', "chainsaw"),
26    ('red', "strawberry"),
27    ('yellow', "peach"),
28    ('yellow', "pear"),
29    ('black', "shotgun"),
30]
31
32class 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
94def test_suite():
95    suite = unittest.TestSuite()
96
97    suite.addTest(unittest.makeSuite(JoinTestCase))
98
99    return suite
100