[Python-checkins] python/dist/src/Lib/bsddb/test test_associate.py,1.4,1.5 test_basics.py,1.6,1.7 test_join.py,1.3,1.4
greg@users.sourceforge.net
greg@users.sourceforge.net
Tue, 08 Jul 2003 21:46:01 -0700
Update of /cvsroot/python/python/dist/src/Lib/bsddb/test
In directory sc8-pr-cvs1:/tmp/cvs-serv26130/bsddb/test
Modified Files:
test_associate.py test_basics.py test_join.py
Log Message:
bsddb 4.1.6:
* Extended DB & DBEnv set_get_returns_none functionality to take a
"level" instead of a boolean flag. The boolean 0 and 1 values still
have the same effect. A value of 2 extends the "return None instead
of raising an exception" behaviour to the DBCursor set methods.
This will become the default behaviour in pybsddb 4.2.
* Fixed a typo in DBCursor.join_item method that made it crash instead
of returning a value. Obviously nobody uses it. Wrote a test case
for join and join_item.
Index: test_associate.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/bsddb/test/test_associate.py,v
retrieving revision 1.4
retrieving revision 1.5
diff -C2 -d -r1.4 -r1.5
*** test_associate.py 28 Jan 2003 17:20:43 -0000 1.4
--- test_associate.py 9 Jul 2003 04:45:59 -0000 1.5
***************
*** 1,4 ****
"""
! TestCases for multi-threaded access to a DB.
"""
--- 1,4 ----
"""
! TestCases for DB.associate.
"""
Index: test_basics.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/bsddb/test/test_basics.py,v
retrieving revision 1.6
retrieving revision 1.7
diff -C2 -d -r1.6 -r1.7
*** test_basics.py 28 Jan 2003 17:20:43 -0000 1.6
--- test_basics.py 9 Jul 2003 04:45:59 -0000 1.7
***************
*** 283,291 ****
#----------------------------------------
! def test03_SimpleCursorStuff(self):
if verbose:
print '\n', '-=' * 30
! print "Running %s.test03_SimpleCursorStuff..." % \
! self.__class__.__name__
if self.env and self.dbopenflags & db.DB_AUTO_COMMIT:
--- 283,291 ----
#----------------------------------------
! def test03_SimpleCursorStuff(self, get_raises_error=0, set_raises_error=1):
if verbose:
print '\n', '-=' * 30
! print "Running %s.test03_SimpleCursorStuff (get_error %s, set_error %s)..." % \
! (self.__class__.__name__, get_raises_error, set_raises_error)
if self.env and self.dbopenflags & db.DB_AUTO_COMMIT:
***************
*** 301,305 ****
if verbose and count % 100 == 0:
print rec
! rec = c.next()
assert count == 1000
--- 301,313 ----
if verbose and count % 100 == 0:
print rec
! try:
! rec = c.next()
! except db.DBNotFoundError, val:
! if get_raises_error:
! assert val[0] == db.DB_NOTFOUND
! if verbose: print val
! rec = None
! else:
! self.fail("unexpected DBNotFoundError")
assert count == 1000
***************
*** 312,316 ****
if verbose and count % 100 == 0:
print rec
! rec = c.prev()
assert count == 1000
--- 320,332 ----
if verbose and count % 100 == 0:
print rec
! try:
! rec = c.prev()
! except db.DBNotFoundError, val:
! if get_raises_error:
! assert val[0] == db.DB_NOTFOUND
! if verbose: print val
! rec = None
! else:
! self.fail("unexpected DBNotFoundError")
assert count == 1000
***************
*** 323,332 ****
try:
! c.set('bad key')
except db.DBNotFoundError, val:
assert val[0] == db.DB_NOTFOUND
if verbose: print val
else:
! self.fail("expected exception")
rec = c.get_both('0404', self.makeData('0404'))
--- 339,351 ----
try:
! n = c.set('bad key')
except db.DBNotFoundError, val:
assert val[0] == db.DB_NOTFOUND
if verbose: print val
else:
! if set_raises_error:
! self.fail("expected exception")
! if n != None:
! self.fail("expected None: "+`n`)
rec = c.get_both('0404', self.makeData('0404'))
***************
*** 334,343 ****
try:
! c.get_both('0404', 'bad data')
except db.DBNotFoundError, val:
assert val[0] == db.DB_NOTFOUND
if verbose: print val
else:
! self.fail("expected exception")
if self.d.get_type() == db.DB_BTREE:
--- 353,365 ----
try:
! n = c.get_both('0404', 'bad data')
except db.DBNotFoundError, val:
assert val[0] == db.DB_NOTFOUND
if verbose: print val
else:
! if get_raises_error:
! self.fail("expected exception")
! if n != None:
! self.fail("expected None: "+`n`)
if self.d.get_type() == db.DB_BTREE:
***************
*** 415,418 ****
--- 437,463 ----
del oldcursor
+ def test03b_SimpleCursorWithoutGetReturnsNone0(self):
+ # same test but raise exceptions instead of returning None
+ if verbose:
+ print '\n', '-=' * 30
+ print "Running %s.test03b_SimpleCursorStuffWithoutGetReturnsNone..." % \
+ self.__class__.__name__
+
+ old = self.d.set_get_returns_none(0)
+ assert old == 1
+ self.test03_SimpleCursorStuff(get_raises_error=1, set_raises_error=1)
+
+ def test03c_SimpleCursorGetReturnsNone2(self):
+ # same test but raise exceptions instead of returning None
+ if verbose:
+ print '\n', '-=' * 30
+ print "Running %s.test03c_SimpleCursorStuffWithoutSetReturnsNone..." % \
+ self.__class__.__name__
+
+ old = self.d.set_get_returns_none(2)
+ assert old == 1
+ old = self.d.set_get_returns_none(2)
+ assert old == 2
+ self.test03_SimpleCursorStuff(get_raises_error=0, set_raises_error=0)
#----------------------------------------
Index: test_join.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/bsddb/test/test_join.py,v
retrieving revision 1.3
retrieving revision 1.4
diff -C2 -d -r1.3 -r1.4
*** test_join.py 30 Dec 2002 20:44:16 -0000 1.3
--- test_join.py 9 Jul 2003 04:45:59 -0000 1.4
***************
*** 2,9 ****
--- 2,117 ----
"""
+ import sys, os, string
+ import tempfile
+ import time
+ from pprint import pprint
+
+ try:
+ from threading import Thread, currentThread
+ have_threads = 1
+ except ImportError:
+ have_threads = 0
+
import unittest
+ from test_all import verbose
+
+ try:
+ # For Python 2.3
+ from bsddb import db, dbshelve
+ except ImportError:
+ # For earlier Pythons w/distutils pybsddb
+ from bsddb3 import db, dbshelve
+
+
+ #----------------------------------------------------------------------
+
+ ProductIndex = [
+ ('apple', "Convenience Store"),
+ ('blueberry', "Farmer's Market"),
+ ('shotgun', "S-Mart"), # Aisle 12
+ ('pear', "Farmer's Market"),
+ ('chainsaw', "S-Mart"), # "Shop smart. Shop S-Mart!"
+ ('strawberry', "Farmer's Market"),
+ ]
+
+ ColorIndex = [
+ ('blue', "blueberry"),
+ ('red', "apple"),
+ ('red', "chainsaw"),
+ ('red', "strawberry"),
+ ('yellow', "peach"),
+ ('yellow', "pear"),
+ ('black', "shotgun"),
+ ]
+
+ class JoinTestCase(unittest.TestCase):
+ keytype = ''
+
+ def setUp(self):
+ self.filename = self.__class__.__name__ + '.db'
+ homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home')
+ self.homeDir = homeDir
+ try: os.mkdir(homeDir)
+ except os.error: pass
+ self.env = db.DBEnv()
+ self.env.open(homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOCK )
+
+ def tearDown(self):
+ self.env.close()
+ import glob
+ files = glob.glob(os.path.join(self.homeDir, '*'))
+ for file in files:
+ os.remove(file)
+
+ def test01_join(self):
+ if verbose:
+ print '\n', '-=' * 30
+ print "Running %s.test01_join..." % \
+ self.__class__.__name__
+
+ # create and populate primary index
+ priDB = db.DB(self.env)
+ priDB.open(self.filename, "primary", db.DB_BTREE, db.DB_CREATE)
+ map(lambda t: apply(priDB.put, t), ProductIndex)
+
+ # create and populate secondary index
+ secDB = db.DB(self.env)
+ secDB.set_flags(db.DB_DUP | db.DB_DUPSORT)
+ secDB.open(self.filename, "secondary", db.DB_BTREE, db.DB_CREATE)
+ map(lambda t: apply(secDB.put, t), ColorIndex)
+
+ sCursor = None
+ jCursor = None
+ try:
+ # lets look up all of the red Products
+ sCursor = secDB.cursor()
+ assert sCursor.set('red')
+
+ # FIXME: jCursor doesn't properly hold a reference to its
+ # cursors, if they are closed before jcursor is used it
+ # can cause a crash.
+ jCursor = priDB.join([sCursor])
+
+ if jCursor.get(0) != ('apple', "Convenience Store"):
+ self.fail("join cursor positioned wrong")
+ if jCursor.join_item() != 'chainsaw':
+ self.fail("DBCursor.join_item returned wrong item")
+ if jCursor.get(0)[0] != 'strawberry':
+ self.fail("join cursor returned wrong thing")
+ if jCursor.get(0): # there were only three red items to return
+ self.fail("join cursor returned too many items")
+ finally:
+ if jCursor:
+ jCursor.close()
+ if sCursor:
+ sCursor.close()
+ priDB.close()
+ secDB.close()
def test_suite():
suite = unittest.TestSuite()
+
+ suite.addTest(unittest.makeSuite(JoinTestCase))
+
return suite