summaryrefslogtreecommitdiff
path: root/Lib/bsddb/test
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/bsddb/test')
-rw-r--r--Lib/bsddb/test/test_all.py9
-rw-r--r--Lib/bsddb/test/test_basics.py50
-rw-r--r--Lib/bsddb/test/test_compare.py394
-rw-r--r--Lib/bsddb/test/test_db.py145
-rw-r--r--Lib/bsddb/test/test_dbenv.py70
-rw-r--r--Lib/bsddb/test/test_dbshelve.py7
-rw-r--r--Lib/bsddb/test/test_dbtables.py2
-rw-r--r--Lib/bsddb/test/test_distributed_transactions.py21
-rw-r--r--Lib/bsddb/test/test_early_close.py21
-rw-r--r--Lib/bsddb/test/test_lock.py7
-rw-r--r--Lib/bsddb/test/test_misc.py11
-rw-r--r--Lib/bsddb/test/test_queue.py5
-rw-r--r--Lib/bsddb/test/test_recno.py26
-rw-r--r--Lib/bsddb/test/test_replication.py86
-rw-r--r--Lib/bsddb/test/test_sequence.py8
-rw-r--r--Lib/bsddb/test/test_thread.py4
16 files changed, 522 insertions, 344 deletions
diff --git a/Lib/bsddb/test/test_all.py b/Lib/bsddb/test/test_all.py
index e9fe618..caef1ac 100644
--- a/Lib/bsddb/test/test_all.py
+++ b/Lib/bsddb/test/test_all.py
@@ -392,10 +392,8 @@ if sys.version_info[0] >= 3 :
return self._dbenv.get_tmp_dir().decode(charset)
def get_data_dirs(self) :
- # Have to use a list comprehension and not
- # generators, because we are supporting Python 2.3.
return tuple(
- [i.decode(charset) for i in self._dbenv.get_data_dirs()])
+ (i.decode(charset) for i in self._dbenv.get_data_dirs()))
class DBSequence_py3k(object) :
def __init__(self, db, *args, **kwargs) :
@@ -484,6 +482,8 @@ def print_versions():
print '-=' * 38
print db.DB_VERSION_STRING
print 'bsddb.db.version(): %s' % (db.version(), )
+ if db.version() >= (5, 0) :
+ print 'bsddb.db.full_version(): %s' %repr(db.full_version())
print 'bsddb.db.__version__: %s' % db.__version__
print 'bsddb.db.cvsid: %s' % db.cvsid
@@ -528,7 +528,8 @@ def get_new_database_path() :
# This path can be overriden via "set_test_path_prefix()".
import os, os.path
-get_new_path.prefix=os.path.join(os.sep,"tmp","z-Berkeley_DB")
+get_new_path.prefix=os.path.join(os.environ.get("TMPDIR",
+ os.path.join(os.sep,"tmp")), "z-Berkeley_DB")
get_new_path.num=0
def get_test_path_prefix() :
diff --git a/Lib/bsddb/test/test_basics.py b/Lib/bsddb/test/test_basics.py
index 4d43390..3c57be4 100644
--- a/Lib/bsddb/test/test_basics.py
+++ b/Lib/bsddb/test/test_basics.py
@@ -9,6 +9,7 @@ import string
from pprint import pprint
import unittest
import time
+import sys
from test_all import db, test_support, verbose, get_new_environment_path, \
get_new_database_path
@@ -44,13 +45,6 @@ class BasicTestCase(unittest.TestCase):
_numKeys = 1002 # PRIVATE. NOTE: must be an even value
- import sys
- if sys.version_info < (2, 4):
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
- def assertFalse(self, expr, msg=None):
- self.failIf(expr,msg=msg)
-
def setUp(self):
if self.useEnv:
self.homeDir=get_new_environment_path()
@@ -74,14 +68,13 @@ class BasicTestCase(unittest.TestCase):
# create and open the DB
self.d = db.DB(self.env)
if not self.useEnv :
- if db.version() >= (4, 2) :
- self.d.set_cachesize(*self.cachesize)
- cachesize = self.d.get_cachesize()
- self.assertEqual(cachesize[0], self.cachesize[0])
- self.assertEqual(cachesize[2], self.cachesize[2])
- # Berkeley DB expands the cache 25% accounting overhead,
- # if the cache is small.
- self.assertEqual(125, int(100.0*cachesize[1]/self.cachesize[1]))
+ self.d.set_cachesize(*self.cachesize)
+ cachesize = self.d.get_cachesize()
+ self.assertEqual(cachesize[0], self.cachesize[0])
+ self.assertEqual(cachesize[2], self.cachesize[2])
+ # Berkeley DB expands the cache 25% accounting overhead,
+ # if the cache is small.
+ self.assertEqual(125, int(100.0*cachesize[1]/self.cachesize[1]))
self.d.set_flags(self.dbsetflags)
if self.dbname:
self.d.open(self.filename, self.dbname, self.dbtype,
@@ -161,7 +154,6 @@ class BasicTestCase(unittest.TestCase):
try:
d.delete('abcd')
except db.DBNotFoundError, val:
- import sys
if sys.version_info < (2, 6) :
self.assertEqual(val[0], db.DB_NOTFOUND)
else :
@@ -184,7 +176,6 @@ class BasicTestCase(unittest.TestCase):
try:
d.put('abcd', 'this should fail', flags=db.DB_NOOVERWRITE)
except db.DBKeyExistError, val:
- import sys
if sys.version_info < (2, 6) :
self.assertEqual(val[0], db.DB_KEYEXIST)
else :
@@ -338,7 +329,6 @@ class BasicTestCase(unittest.TestCase):
rec = c.next()
except db.DBNotFoundError, val:
if get_raises_error:
- import sys
if sys.version_info < (2, 6) :
self.assertEqual(val[0], db.DB_NOTFOUND)
else :
@@ -363,7 +353,6 @@ class BasicTestCase(unittest.TestCase):
rec = c.prev()
except db.DBNotFoundError, val:
if get_raises_error:
- import sys
if sys.version_info < (2, 6) :
self.assertEqual(val[0], db.DB_NOTFOUND)
else :
@@ -390,7 +379,6 @@ class BasicTestCase(unittest.TestCase):
try:
n = c.set('bad key')
except db.DBNotFoundError, val:
- import sys
if sys.version_info < (2, 6) :
self.assertEqual(val[0], db.DB_NOTFOUND)
else :
@@ -408,7 +396,6 @@ class BasicTestCase(unittest.TestCase):
try:
n = c.get_both('0404', 'bad data')
except db.DBNotFoundError, val:
- import sys
if sys.version_info < (2, 6) :
self.assertEqual(val[0], db.DB_NOTFOUND)
else :
@@ -441,7 +428,6 @@ class BasicTestCase(unittest.TestCase):
rec = c.current()
except db.DBKeyEmptyError, val:
if get_raises_error:
- import sys
if sys.version_info < (2, 6) :
self.assertEqual(val[0], db.DB_KEYEMPTY)
else :
@@ -490,7 +476,6 @@ class BasicTestCase(unittest.TestCase):
# a bug may cause a NULL pointer dereference...
getattr(c, method)(*args)
except db.DBError, val:
- import sys
if sys.version_info < (2, 6) :
self.assertEqual(val[0], 0)
else :
@@ -712,11 +697,6 @@ class BasicHashWithEnvTestCase(BasicWithEnvTestCase):
#----------------------------------------------------------------------
class BasicTransactionTestCase(BasicTestCase):
- import sys
- if sys.version_info < (2, 4):
- def assertTrue(self, expr, msg=None):
- return self.failUnless(expr,msg=msg)
-
if (sys.version_info < (2, 7)) or ((sys.version_info >= (3, 0)) and
(sys.version_info < (3, 2))) :
def assertIn(self, a, b, msg=None) :
@@ -792,7 +772,6 @@ class BasicTransactionTestCase(BasicTestCase):
for log in logs:
if verbose:
print 'log file: ' + log
- if db.version() >= (4,2):
logs = self.env.log_archive(db.DB_ARCH_REMOVE)
self.assertTrue(not logs)
@@ -875,7 +854,6 @@ class BasicTransactionTestCase(BasicTestCase):
#----------------------------------------
- if db.version() >= (4, 2) :
def test_get_tx_max(self) :
self.assertEqual(self.env.get_tx_max(), 30)
@@ -1098,11 +1076,6 @@ class HashMultiDBTestCase(BasicMultiDBTestCase):
class PrivateObject(unittest.TestCase) :
- import sys
- if sys.version_info < (2, 4):
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
def tearDown(self) :
del self.obj
@@ -1116,7 +1089,6 @@ class PrivateObject(unittest.TestCase) :
self.assertTrue(a is b) # Object identity
def test03_leak_assignment(self) :
- import sys
a = "example of private object"
refcount = sys.getrefcount(a)
self.obj.set_private(a)
@@ -1125,7 +1097,6 @@ class PrivateObject(unittest.TestCase) :
self.assertEqual(refcount, sys.getrefcount(a))
def test04_leak_GC(self) :
- import sys
a = "example of private object"
refcount = sys.getrefcount(a)
self.obj.set_private(a)
@@ -1141,11 +1112,6 @@ class DBPrivateObject(PrivateObject) :
self.obj = db.DB()
class CrashAndBurn(unittest.TestCase) :
- import sys
- if sys.version_info < (2, 4):
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
#def test01_OpenCrash(self) :
# # See http://bugs.python.org/issue3307
# self.assertRaises(db.DBInvalidArgError, db.DB, None, 65535)
diff --git a/Lib/bsddb/test/test_compare.py b/Lib/bsddb/test/test_compare.py
index db9cd74..cb3b463 100644
--- a/Lib/bsddb/test/test_compare.py
+++ b/Lib/bsddb/test/test_compare.py
@@ -1,5 +1,5 @@
"""
-TestCases for python DB Btree key comparison function.
+TestCases for python DB duplicate and Btree key comparison function.
"""
import sys, os, re
@@ -20,31 +20,24 @@ def cmp(a, b) :
lexical_cmp = cmp
-def lowercase_cmp(left, right):
- return cmp (left.lower(), right.lower())
+def lowercase_cmp(left, right) :
+ return cmp(left.lower(), right.lower())
-def make_reverse_comparator (cmp):
- def reverse (left, right, delegate=cmp):
- return - delegate (left, right)
+def make_reverse_comparator(cmp) :
+ def reverse(left, right, delegate=cmp) :
+ return - delegate(left, right)
return reverse
_expected_lexical_test_data = ['', 'CCCP', 'a', 'aaa', 'b', 'c', 'cccce', 'ccccf']
_expected_lowercase_test_data = ['', 'a', 'aaa', 'b', 'c', 'CC', 'cccce', 'ccccf', 'CCCP']
-class ComparatorTests (unittest.TestCase):
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None) :
- return self.failUnless(expr,msg=msg)
-
- def comparator_test_helper (self, comparator, expected_data):
+class ComparatorTests(unittest.TestCase) :
+ def comparator_test_helper(self, comparator, expected_data) :
data = expected_data[:]
import sys
if sys.version_info < (2, 6) :
- if sys.version_info < (2, 4) :
- data.sort(comparator)
- else :
- data.sort(cmp=comparator)
+ data.sort(cmp=comparator)
else : # Insertion Sort. Please, improve
data2 = []
for i in data :
@@ -60,143 +53,139 @@ class ComparatorTests (unittest.TestCase):
self.assertEqual(data, expected_data,
"comparator `%s' is not right: %s vs. %s"
% (comparator, expected_data, data))
- def test_lexical_comparator (self):
- self.comparator_test_helper (lexical_cmp, _expected_lexical_test_data)
- def test_reverse_lexical_comparator (self):
+ def test_lexical_comparator(self) :
+ self.comparator_test_helper(lexical_cmp, _expected_lexical_test_data)
+ def test_reverse_lexical_comparator(self) :
rev = _expected_lexical_test_data[:]
- rev.reverse ()
- self.comparator_test_helper (make_reverse_comparator (lexical_cmp),
+ rev.reverse()
+ self.comparator_test_helper(make_reverse_comparator(lexical_cmp),
rev)
- def test_lowercase_comparator (self):
- self.comparator_test_helper (lowercase_cmp,
+ def test_lowercase_comparator(self) :
+ self.comparator_test_helper(lowercase_cmp,
_expected_lowercase_test_data)
-class AbstractBtreeKeyCompareTestCase (unittest.TestCase):
+class AbstractBtreeKeyCompareTestCase(unittest.TestCase) :
env = None
db = None
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
if (sys.version_info < (2, 7)) or ((sys.version_info >= (3,0)) and
(sys.version_info < (3, 2))) :
def assertLess(self, a, b, msg=None) :
return self.assertTrue(a<b, msg=msg)
- def setUp (self):
+ def setUp(self) :
self.filename = self.__class__.__name__ + '.db'
self.homeDir = get_new_environment_path()
env = db.DBEnv()
- env.open (self.homeDir,
+ env.open(self.homeDir,
db.DB_CREATE | db.DB_INIT_MPOOL
| db.DB_INIT_LOCK | db.DB_THREAD)
self.env = env
- def tearDown (self):
+ def tearDown(self) :
self.closeDB()
if self.env is not None:
self.env.close()
self.env = None
test_support.rmtree(self.homeDir)
- def addDataToDB (self, data):
+ def addDataToDB(self, data) :
i = 0
for item in data:
- self.db.put (item, str (i))
+ self.db.put(item, str(i))
i = i + 1
- def createDB (self, key_comparator):
- self.db = db.DB (self.env)
- self.setupDB (key_comparator)
- self.db.open (self.filename, "test", db.DB_BTREE, db.DB_CREATE)
+ def createDB(self, key_comparator) :
+ self.db = db.DB(self.env)
+ self.setupDB(key_comparator)
+ self.db.open(self.filename, "test", db.DB_BTREE, db.DB_CREATE)
- def setupDB (self, key_comparator):
- self.db.set_bt_compare (key_comparator)
+ def setupDB(self, key_comparator) :
+ self.db.set_bt_compare(key_comparator)
- def closeDB (self):
+ def closeDB(self) :
if self.db is not None:
- self.db.close ()
+ self.db.close()
self.db = None
- def startTest (self):
+ def startTest(self) :
pass
- def finishTest (self, expected = None):
+ def finishTest(self, expected = None) :
if expected is not None:
- self.check_results (expected)
- self.closeDB ()
+ self.check_results(expected)
+ self.closeDB()
- def check_results (self, expected):
- curs = self.db.cursor ()
+ def check_results(self, expected) :
+ curs = self.db.cursor()
try:
index = 0
- rec = curs.first ()
+ rec = curs.first()
while rec:
key, ignore = rec
- self.assertLess(index, len (expected),
+ self.assertLess(index, len(expected),
"to many values returned from cursor")
self.assertEqual(expected[index], key,
"expected value `%s' at %d but got `%s'"
% (expected[index], index, key))
index = index + 1
- rec = curs.next ()
- self.assertEqual(index, len (expected),
+ rec = curs.next()
+ self.assertEqual(index, len(expected),
"not enough values returned from cursor")
finally:
- curs.close ()
+ curs.close()
-class BtreeKeyCompareTestCase (AbstractBtreeKeyCompareTestCase):
- def runCompareTest (self, comparator, data):
- self.startTest ()
- self.createDB (comparator)
- self.addDataToDB (data)
- self.finishTest (data)
+class BtreeKeyCompareTestCase(AbstractBtreeKeyCompareTestCase) :
+ def runCompareTest(self, comparator, data) :
+ self.startTest()
+ self.createDB(comparator)
+ self.addDataToDB(data)
+ self.finishTest(data)
- def test_lexical_ordering (self):
- self.runCompareTest (lexical_cmp, _expected_lexical_test_data)
+ def test_lexical_ordering(self) :
+ self.runCompareTest(lexical_cmp, _expected_lexical_test_data)
- def test_reverse_lexical_ordering (self):
+ def test_reverse_lexical_ordering(self) :
expected_rev_data = _expected_lexical_test_data[:]
- expected_rev_data.reverse ()
- self.runCompareTest (make_reverse_comparator (lexical_cmp),
+ expected_rev_data.reverse()
+ self.runCompareTest(make_reverse_comparator(lexical_cmp),
expected_rev_data)
- def test_compare_function_useless (self):
- self.startTest ()
- def socialist_comparator (l, r):
+ def test_compare_function_useless(self) :
+ self.startTest()
+ def socialist_comparator(l, r) :
return 0
- self.createDB (socialist_comparator)
- self.addDataToDB (['b', 'a', 'd'])
+ self.createDB(socialist_comparator)
+ self.addDataToDB(['b', 'a', 'd'])
# all things being equal the first key will be the only key
# in the database... (with the last key's value fwiw)
- self.finishTest (['b'])
+ self.finishTest(['b'])
-class BtreeExceptionsTestCase (AbstractBtreeKeyCompareTestCase):
- def test_raises_non_callable (self):
- self.startTest ()
- self.assertRaises (TypeError, self.createDB, 'abc')
- self.assertRaises (TypeError, self.createDB, None)
- self.finishTest ()
+class BtreeExceptionsTestCase(AbstractBtreeKeyCompareTestCase) :
+ def test_raises_non_callable(self) :
+ self.startTest()
+ self.assertRaises(TypeError, self.createDB, 'abc')
+ self.assertRaises(TypeError, self.createDB, None)
+ self.finishTest()
- def test_set_bt_compare_with_function (self):
- self.startTest ()
- self.createDB (lexical_cmp)
- self.finishTest ()
+ def test_set_bt_compare_with_function(self) :
+ self.startTest()
+ self.createDB(lexical_cmp)
+ self.finishTest()
- def check_results (self, results):
+ def check_results(self, results) :
pass
- def test_compare_function_incorrect (self):
- self.startTest ()
- def bad_comparator (l, r):
+ def test_compare_function_incorrect(self) :
+ self.startTest()
+ def bad_comparator(l, r) :
return 1
# verify that set_bt_compare checks that comparator('', '') == 0
- self.assertRaises (TypeError, self.createDB, bad_comparator)
- self.finishTest ()
+ self.assertRaises(TypeError, self.createDB, bad_comparator)
+ self.finishTest()
- def verifyStderr(self, method, successRe):
+ def verifyStderr(self, method, successRe) :
"""
Call method() while capturing sys.stderr output internally and
call self.fail() if successRe.search() does not match the stderr
@@ -210,64 +199,249 @@ class BtreeExceptionsTestCase (AbstractBtreeKeyCompareTestCase):
temp = sys.stderr
sys.stderr = stdErr
errorOut = temp.getvalue()
- if not successRe.search(errorOut):
+ if not successRe.search(errorOut) :
self.fail("unexpected stderr output:\n"+errorOut)
if sys.version_info < (3, 0) : # XXX: How to do this in Py3k ???
sys.exc_traceback = sys.last_traceback = None
- def _test_compare_function_exception (self):
- self.startTest ()
- def bad_comparator (l, r):
+ def _test_compare_function_exception(self) :
+ self.startTest()
+ def bad_comparator(l, r) :
if l == r:
# pass the set_bt_compare test
return 0
raise RuntimeError, "i'm a naughty comparison function"
- self.createDB (bad_comparator)
+ self.createDB(bad_comparator)
#print "\n*** test should print 2 uncatchable tracebacks ***"
- self.addDataToDB (['a', 'b', 'c']) # this should raise, but...
- self.finishTest ()
+ self.addDataToDB(['a', 'b', 'c']) # this should raise, but...
+ self.finishTest()
- def test_compare_function_exception(self):
+ def test_compare_function_exception(self) :
self.verifyStderr(
self._test_compare_function_exception,
re.compile('(^RuntimeError:.* naughty.*){2}', re.M|re.S)
)
- def _test_compare_function_bad_return (self):
- self.startTest ()
- def bad_comparator (l, r):
+ def _test_compare_function_bad_return(self) :
+ self.startTest()
+ def bad_comparator(l, r) :
if l == r:
# pass the set_bt_compare test
return 0
return l
- self.createDB (bad_comparator)
+ self.createDB(bad_comparator)
+ #print "\n*** test should print 2 errors about returning an int ***"
+ self.addDataToDB(['a', 'b', 'c']) # this should raise, but...
+ self.finishTest()
+
+ def test_compare_function_bad_return(self) :
+ self.verifyStderr(
+ self._test_compare_function_bad_return,
+ re.compile('(^TypeError:.* return an int.*){2}', re.M|re.S)
+ )
+
+
+ def test_cannot_assign_twice(self) :
+
+ def my_compare(a, b) :
+ return 0
+
+ self.startTest()
+ self.createDB(my_compare)
+ self.assertRaises(RuntimeError, self.db.set_bt_compare, my_compare)
+
+class AbstractDuplicateCompareTestCase(unittest.TestCase) :
+ env = None
+ db = None
+
+ if (sys.version_info < (2, 7)) or ((sys.version_info >= (3,0)) and
+ (sys.version_info < (3, 2))) :
+ def assertLess(self, a, b, msg=None) :
+ return self.assertTrue(a<b, msg=msg)
+
+ def setUp(self) :
+ self.filename = self.__class__.__name__ + '.db'
+ self.homeDir = get_new_environment_path()
+ env = db.DBEnv()
+ env.open(self.homeDir,
+ db.DB_CREATE | db.DB_INIT_MPOOL
+ | db.DB_INIT_LOCK | db.DB_THREAD)
+ self.env = env
+
+ def tearDown(self) :
+ self.closeDB()
+ if self.env is not None:
+ self.env.close()
+ self.env = None
+ test_support.rmtree(self.homeDir)
+
+ def addDataToDB(self, data) :
+ for item in data:
+ self.db.put("key", item)
+
+ def createDB(self, dup_comparator) :
+ self.db = db.DB(self.env)
+ self.setupDB(dup_comparator)
+ self.db.open(self.filename, "test", db.DB_BTREE, db.DB_CREATE)
+
+ def setupDB(self, dup_comparator) :
+ self.db.set_flags(db.DB_DUPSORT)
+ self.db.set_dup_compare(dup_comparator)
+
+ def closeDB(self) :
+ if self.db is not None:
+ self.db.close()
+ self.db = None
+
+ def startTest(self) :
+ pass
+
+ def finishTest(self, expected = None) :
+ if expected is not None:
+ self.check_results(expected)
+ self.closeDB()
+
+ def check_results(self, expected) :
+ curs = self.db.cursor()
+ try:
+ index = 0
+ rec = curs.first()
+ while rec:
+ ignore, data = rec
+ self.assertLess(index, len(expected),
+ "to many values returned from cursor")
+ self.assertEqual(expected[index], data,
+ "expected value `%s' at %d but got `%s'"
+ % (expected[index], index, data))
+ index = index + 1
+ rec = curs.next()
+ self.assertEqual(index, len(expected),
+ "not enough values returned from cursor")
+ finally:
+ curs.close()
+
+class DuplicateCompareTestCase(AbstractDuplicateCompareTestCase) :
+ def runCompareTest(self, comparator, data) :
+ self.startTest()
+ self.createDB(comparator)
+ self.addDataToDB(data)
+ self.finishTest(data)
+
+ def test_lexical_ordering(self) :
+ self.runCompareTest(lexical_cmp, _expected_lexical_test_data)
+
+ def test_reverse_lexical_ordering(self) :
+ expected_rev_data = _expected_lexical_test_data[:]
+ expected_rev_data.reverse()
+ self.runCompareTest(make_reverse_comparator(lexical_cmp),
+ expected_rev_data)
+
+class DuplicateExceptionsTestCase(AbstractDuplicateCompareTestCase) :
+ def test_raises_non_callable(self) :
+ self.startTest()
+ self.assertRaises(TypeError, self.createDB, 'abc')
+ self.assertRaises(TypeError, self.createDB, None)
+ self.finishTest()
+
+ def test_set_dup_compare_with_function(self) :
+ self.startTest()
+ self.createDB(lexical_cmp)
+ self.finishTest()
+
+ def check_results(self, results) :
+ pass
+
+ def test_compare_function_incorrect(self) :
+ self.startTest()
+ def bad_comparator(l, r) :
+ return 1
+ # verify that set_dup_compare checks that comparator('', '') == 0
+ self.assertRaises(TypeError, self.createDB, bad_comparator)
+ self.finishTest()
+
+ def test_compare_function_useless(self) :
+ self.startTest()
+ def socialist_comparator(l, r) :
+ return 0
+ self.createDB(socialist_comparator)
+ # DUPSORT does not allow "duplicate duplicates"
+ self.assertRaises(db.DBKeyExistError, self.addDataToDB, ['b', 'a', 'd'])
+ self.finishTest()
+
+ def verifyStderr(self, method, successRe) :
+ """
+ Call method() while capturing sys.stderr output internally and
+ call self.fail() if successRe.search() does not match the stderr
+ output. This is used to test for uncatchable exceptions.
+ """
+ stdErr = sys.stderr
+ sys.stderr = StringIO()
+ try:
+ method()
+ finally:
+ temp = sys.stderr
+ sys.stderr = stdErr
+ errorOut = temp.getvalue()
+ if not successRe.search(errorOut) :
+ self.fail("unexpected stderr output:\n"+errorOut)
+ if sys.version_info < (3, 0) : # XXX: How to do this in Py3k ???
+ sys.exc_traceback = sys.last_traceback = None
+
+ def _test_compare_function_exception(self) :
+ self.startTest()
+ def bad_comparator(l, r) :
+ if l == r:
+ # pass the set_dup_compare test
+ return 0
+ raise RuntimeError, "i'm a naughty comparison function"
+ self.createDB(bad_comparator)
+ #print "\n*** test should print 2 uncatchable tracebacks ***"
+ self.addDataToDB(['a', 'b', 'c']) # this should raise, but...
+ self.finishTest()
+
+ def test_compare_function_exception(self) :
+ self.verifyStderr(
+ self._test_compare_function_exception,
+ re.compile('(^RuntimeError:.* naughty.*){2}', re.M|re.S)
+ )
+
+ def _test_compare_function_bad_return(self) :
+ self.startTest()
+ def bad_comparator(l, r) :
+ if l == r:
+ # pass the set_dup_compare test
+ return 0
+ return l
+ self.createDB(bad_comparator)
#print "\n*** test should print 2 errors about returning an int ***"
- self.addDataToDB (['a', 'b', 'c']) # this should raise, but...
- self.finishTest ()
+ self.addDataToDB(['a', 'b', 'c']) # this should raise, but...
+ self.finishTest()
- def test_compare_function_bad_return(self):
+ def test_compare_function_bad_return(self) :
self.verifyStderr(
self._test_compare_function_bad_return,
re.compile('(^TypeError:.* return an int.*){2}', re.M|re.S)
)
- def test_cannot_assign_twice (self):
+ def test_cannot_assign_twice(self) :
- def my_compare (a, b):
+ def my_compare(a, b) :
return 0
self.startTest()
self.createDB(my_compare)
- self.assertRaises (RuntimeError, self.db.set_bt_compare, my_compare)
+ self.assertRaises(RuntimeError, self.db.set_dup_compare, my_compare)
-def test_suite ():
- res = unittest.TestSuite ()
+def test_suite() :
+ res = unittest.TestSuite()
- res.addTest (unittest.makeSuite (ComparatorTests))
- res.addTest (unittest.makeSuite (BtreeExceptionsTestCase))
- res.addTest (unittest.makeSuite (BtreeKeyCompareTestCase))
+ res.addTest(unittest.makeSuite(ComparatorTests))
+ res.addTest(unittest.makeSuite(BtreeExceptionsTestCase))
+ res.addTest(unittest.makeSuite(BtreeKeyCompareTestCase))
+ res.addTest(unittest.makeSuite(DuplicateExceptionsTestCase))
+ res.addTest(unittest.makeSuite(DuplicateCompareTestCase))
return res
if __name__ == '__main__':
- unittest.main (defaultTest = 'suite')
+ unittest.main(defaultTest = 'suite')
diff --git a/Lib/bsddb/test/test_db.py b/Lib/bsddb/test/test_db.py
index 8d3a32b..2bc109f 100644
--- a/Lib/bsddb/test/test_db.py
+++ b/Lib/bsddb/test/test_db.py
@@ -7,11 +7,6 @@ from test_all import db, test_support, get_new_environment_path, \
#----------------------------------------------------------------------
class DB(unittest.TestCase):
- import sys
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
def setUp(self):
self.path = get_new_database_path()
self.db = db.DB()
@@ -19,10 +14,28 @@ class DB(unittest.TestCase):
def tearDown(self):
self.db.close()
del self.db
- test_support.rmtree(self.path)
+ test_support.unlink(self.path)
class DB_general(DB) :
- if db.version() >= (4, 2) :
+ def test_get_open_flags(self) :
+ self.db.open(self.path, dbtype=db.DB_HASH, flags = db.DB_CREATE)
+ self.assertEqual(db.DB_CREATE, self.db.get_open_flags())
+
+ def test_get_open_flags2(self) :
+ self.db.open(self.path, dbtype=db.DB_HASH, flags = db.DB_CREATE |
+ db.DB_THREAD)
+ self.assertEqual(db.DB_CREATE | db.DB_THREAD, self.db.get_open_flags())
+
+ def test_get_dbname_filename(self) :
+ self.db.open(self.path, dbtype=db.DB_HASH, flags = db.DB_CREATE)
+ self.assertEqual((self.path, None), self.db.get_dbname())
+
+ def test_get_dbname_filename_database(self) :
+ name = "jcea-random-name"
+ self.db.open(self.path, dbname=name, dbtype=db.DB_HASH,
+ flags = db.DB_CREATE)
+ self.assertEqual((self.path, name), self.db.get_dbname())
+
def test_bt_minkey(self) :
for i in [17, 108, 1030] :
self.db.set_bt_minkey(i)
@@ -44,30 +57,34 @@ class DB_general(DB) :
self.db.set_priority(flag)
self.assertEqual(flag, self.db.get_priority())
+ def test_get_transactional(self) :
+ self.assertFalse(self.db.get_transactional())
+ self.db.open(self.path, dbtype=db.DB_HASH, flags = db.DB_CREATE)
+ self.assertFalse(self.db.get_transactional())
+
class DB_hash(DB) :
- if db.version() >= (4, 2) :
- def test_h_ffactor(self) :
- for ffactor in [4, 16, 256] :
- self.db.set_h_ffactor(ffactor)
- self.assertEqual(ffactor, self.db.get_h_ffactor())
-
- def test_h_nelem(self) :
- for nelem in [1, 2, 4] :
- nelem = nelem*1024*1024 # Millions
- self.db.set_h_nelem(nelem)
- self.assertEqual(nelem, self.db.get_h_nelem())
-
- def test_pagesize(self) :
- for i in xrange(9, 17) : # From 512 to 65536
- i = 1<<i
- self.db.set_pagesize(i)
- self.assertEqual(i, self.db.get_pagesize())
-
- # The valid values goes from 512 to 65536
- # Test 131072 bytes...
- self.assertRaises(db.DBInvalidArgError, self.db.set_pagesize, 1<<17)
- # Test 256 bytes...
- self.assertRaises(db.DBInvalidArgError, self.db.set_pagesize, 1<<8)
+ def test_h_ffactor(self) :
+ for ffactor in [4, 16, 256] :
+ self.db.set_h_ffactor(ffactor)
+ self.assertEqual(ffactor, self.db.get_h_ffactor())
+
+ def test_h_nelem(self) :
+ for nelem in [1, 2, 4] :
+ nelem = nelem*1024*1024 # Millions
+ self.db.set_h_nelem(nelem)
+ self.assertEqual(nelem, self.db.get_h_nelem())
+
+ def test_pagesize(self) :
+ for i in xrange(9, 17) : # From 512 to 65536
+ i = 1<<i
+ self.db.set_pagesize(i)
+ self.assertEqual(i, self.db.get_pagesize())
+
+ # The valid values goes from 512 to 65536
+ # Test 131072 bytes...
+ self.assertRaises(db.DBInvalidArgError, self.db.set_pagesize, 1<<17)
+ # Test 256 bytes...
+ self.assertRaises(db.DBInvalidArgError, self.db.set_pagesize, 1<<8)
class DB_txn(DB) :
def setUp(self) :
@@ -84,7 +101,6 @@ class DB_txn(DB) :
del self.env
test_support.rmtree(self.homeDir)
- if db.version() >= (4, 2) :
def test_flags(self) :
self.db.set_flags(db.DB_CHKSUM)
self.assertEqual(db.DB_CHKSUM, self.db.get_flags())
@@ -92,40 +108,45 @@ class DB_txn(DB) :
self.assertEqual(db.DB_TXN_NOT_DURABLE | db.DB_CHKSUM,
self.db.get_flags())
+ def test_get_transactional(self) :
+ self.assertFalse(self.db.get_transactional())
+ # DB_AUTO_COMMIT = Implicit transaction
+ self.db.open("XXX", dbtype=db.DB_HASH,
+ flags = db.DB_CREATE | db.DB_AUTO_COMMIT)
+ self.assertTrue(self.db.get_transactional())
+
class DB_recno(DB) :
- if db.version() >= (4, 2) :
- def test_re_pad(self) :
- for i in [' ', '*'] : # Check chars
- self.db.set_re_pad(i)
- self.assertEqual(ord(i), self.db.get_re_pad())
- for i in [97, 65] : # Check integers
- self.db.set_re_pad(i)
- self.assertEqual(i, self.db.get_re_pad())
-
- def test_re_delim(self) :
- for i in [' ', '*'] : # Check chars
- self.db.set_re_delim(i)
- self.assertEqual(ord(i), self.db.get_re_delim())
- for i in [97, 65] : # Check integers
- self.db.set_re_delim(i)
- self.assertEqual(i, self.db.get_re_delim())
-
- def test_re_source(self) :
- for i in ["test", "test2", "test3"] :
- self.db.set_re_source(i)
- self.assertEqual(i, self.db.get_re_source())
+ def test_re_pad(self) :
+ for i in [' ', '*'] : # Check chars
+ self.db.set_re_pad(i)
+ self.assertEqual(ord(i), self.db.get_re_pad())
+ for i in [97, 65] : # Check integers
+ self.db.set_re_pad(i)
+ self.assertEqual(i, self.db.get_re_pad())
+
+ def test_re_delim(self) :
+ for i in [' ', '*'] : # Check chars
+ self.db.set_re_delim(i)
+ self.assertEqual(ord(i), self.db.get_re_delim())
+ for i in [97, 65] : # Check integers
+ self.db.set_re_delim(i)
+ self.assertEqual(i, self.db.get_re_delim())
+
+ def test_re_source(self) :
+ for i in ["test", "test2", "test3"] :
+ self.db.set_re_source(i)
+ self.assertEqual(i, self.db.get_re_source())
class DB_queue(DB) :
- if db.version() >= (4, 2) :
- def test_re_len(self) :
- for i in [33, 65, 300, 2000] :
- self.db.set_re_len(i)
- self.assertEqual(i, self.db.get_re_len())
-
- def test_q_extentsize(self) :
- for i in [1, 60, 100] :
- self.db.set_q_extentsize(i)
- self.assertEqual(i, self.db.get_q_extentsize())
+ def test_re_len(self) :
+ for i in [33, 65, 300, 2000] :
+ self.db.set_re_len(i)
+ self.assertEqual(i, self.db.get_re_len())
+
+ def test_q_extentsize(self) :
+ for i in [1, 60, 100] :
+ self.db.set_q_extentsize(i)
+ self.assertEqual(i, self.db.get_q_extentsize())
def test_suite():
suite = unittest.TestSuite()
diff --git a/Lib/bsddb/test/test_dbenv.py b/Lib/bsddb/test/test_dbenv.py
index 37281df..76ef7db 100644
--- a/Lib/bsddb/test/test_dbenv.py
+++ b/Lib/bsddb/test/test_dbenv.py
@@ -7,14 +7,6 @@ from test_all import db, test_support, get_new_environment_path, \
#----------------------------------------------------------------------
class DBEnv(unittest.TestCase):
- import sys
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
- def assertFalse(self, expr, msg=None):
- self.failIf(expr,msg=msg)
-
def setUp(self):
self.homeDir = get_new_environment_path()
self.env = db.DBEnv()
@@ -25,12 +17,31 @@ class DBEnv(unittest.TestCase):
test_support.rmtree(self.homeDir)
class DBEnv_general(DBEnv) :
+ def test_get_open_flags(self) :
+ flags = db.DB_CREATE | db.DB_INIT_MPOOL
+ self.env.open(self.homeDir, flags)
+ self.assertEqual(flags, self.env.get_open_flags())
+
+ def test_get_open_flags2(self) :
+ flags = db.DB_CREATE | db.DB_INIT_MPOOL | \
+ db.DB_INIT_LOCK | db.DB_THREAD
+ self.env.open(self.homeDir, flags)
+ self.assertEqual(flags, self.env.get_open_flags())
+
if db.version() >= (4, 7) :
def test_lk_partitions(self) :
for i in [10, 20, 40] :
self.env.set_lk_partitions(i)
self.assertEqual(i, self.env.get_lk_partitions())
+ def test_getset_intermediate_dir_mode(self) :
+ self.assertEqual(None, self.env.get_intermediate_dir_mode())
+ for mode in ["rwx------", "rw-rw-rw-", "rw-r--r--"] :
+ self.env.set_intermediate_dir_mode(mode)
+ self.assertEqual(mode, self.env.get_intermediate_dir_mode())
+ self.assertRaises(db.DBInvalidArgError,
+ self.env.set_intermediate_dir_mode, "abcde")
+
if db.version() >= (4, 6) :
def test_thread(self) :
for i in [16, 100, 1000] :
@@ -58,21 +69,19 @@ class DBEnv_general(DBEnv) :
self.env.set_lg_filemode(i)
self.assertEqual(i, self.env.get_lg_filemode())
- if db.version() >= (4, 3) :
- def test_mp_max_openfd(self) :
- for i in [17, 31, 42] :
- self.env.set_mp_max_openfd(i)
- self.assertEqual(i, self.env.get_mp_max_openfd())
-
- def test_mp_max_write(self) :
- for i in [100, 200, 300] :
- for j in [1, 2, 3] :
- j *= 1000000
- self.env.set_mp_max_write(i, j)
- v=self.env.get_mp_max_write()
- self.assertEqual((i, j), v)
-
- if db.version() >= (4, 2) :
+ def test_mp_max_openfd(self) :
+ for i in [17, 31, 42] :
+ self.env.set_mp_max_openfd(i)
+ self.assertEqual(i, self.env.get_mp_max_openfd())
+
+ def test_mp_max_write(self) :
+ for i in [100, 200, 300] :
+ for j in [1, 2, 3] :
+ j *= 1000000
+ self.env.set_mp_max_write(i, j)
+ v=self.env.get_mp_max_write()
+ self.assertEqual((i, j), v)
+
def test_invalid_txn(self) :
# This environment doesn't support transactions
self.assertRaises(db.DBInvalidArgError, self.env.txn_begin)
@@ -115,7 +124,7 @@ class DBEnv_general(DBEnv) :
self.assertEqual(i, self.env.get_lk_max_lockers())
def test_lg_regionmax(self) :
- for i in [128, 256, 1024] :
+ for i in [128, 256, 1000] :
i = i*1024*1024
self.env.set_lg_regionmax(i)
j = self.env.get_lg_regionmax()
@@ -127,8 +136,7 @@ class DBEnv_general(DBEnv) :
db.DB_LOCK_MINLOCKS, db.DB_LOCK_MINWRITE,
db.DB_LOCK_OLDEST, db.DB_LOCK_RANDOM, db.DB_LOCK_YOUNGEST]
- if db.version() >= (4, 3) :
- flags.append(db.DB_LOCK_MAXWRITE)
+ flags.append(db.DB_LOCK_MAXWRITE)
for i in flags :
self.env.set_lk_detect(i)
@@ -172,8 +180,12 @@ class DBEnv_general(DBEnv) :
self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL)
cachesize = (0, 2*1024*1024, 1)
self.assertRaises(db.DBInvalidArgError,
- self.env.set_cachesize, *cachesize)
- self.assertEqual(cachesize2, self.env.get_cachesize())
+ self.env.set_cachesize, *cachesize)
+ cachesize3 = self.env.get_cachesize()
+ self.assertEqual(cachesize2[0], cachesize3[0])
+ self.assertEqual(cachesize2[2], cachesize3[2])
+ # In Berkeley DB 5.1, the cachesize can change when opening the Env
+ self.assertTrue(cachesize2[1] <= cachesize3[1])
def test_set_cachesize_dbenv_db(self) :
# You can not configure the cachesize using
@@ -305,7 +317,7 @@ class DBEnv_log_txn(DBEnv) :
self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL |
db.DB_INIT_LOG | db.DB_INIT_TXN)
- if db.version() >= (4, 5) :
+ if (db.version() >= (4, 5)) and (db.version() < (5, 2)) :
def test_tx_max(self) :
txns=[]
def tx() :
diff --git a/Lib/bsddb/test/test_dbshelve.py b/Lib/bsddb/test/test_dbshelve.py
index c0135fe..c3701e1 100644
--- a/Lib/bsddb/test/test_dbshelve.py
+++ b/Lib/bsddb/test/test_dbshelve.py
@@ -12,9 +12,6 @@ from test_all import db, dbshelve, test_support, verbose, \
-if sys.version_info < (2, 4) :
- from sets import Set as set
-
#----------------------------------------------------------------------
@@ -33,10 +30,6 @@ class DataClass:
class DBShelveTestCase(unittest.TestCase):
- if sys.version_info < (2, 4):
- def assertTrue(self, expr, msg=None):
- return self.failUnless(expr,msg=msg)
-
if (sys.version_info < (2, 7)) or ((sys.version_info >= (3, 0)) and
(sys.version_info < (3, 2))) :
def assertIn(self, a, b, msg=None) :
diff --git a/Lib/bsddb/test/test_dbtables.py b/Lib/bsddb/test/test_dbtables.py
index bb31381..250c492 100644
--- a/Lib/bsddb/test/test_dbtables.py
+++ b/Lib/bsddb/test/test_dbtables.py
@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-#
#-----------------------------------------------------------------------
# A test suite for the table interface built on bsddb.db
#-----------------------------------------------------------------------
diff --git a/Lib/bsddb/test/test_distributed_transactions.py b/Lib/bsddb/test/test_distributed_transactions.py
index 1711fc5..9058575 100644
--- a/Lib/bsddb/test/test_distributed_transactions.py
+++ b/Lib/bsddb/test/test_distributed_transactions.py
@@ -7,13 +7,6 @@ import unittest
from test_all import db, test_support, get_new_environment_path, \
get_new_database_path
-try :
- a=set()
-except : # Python 2.3
- from sets import Set as set
-else :
- del a
-
from test_all import verbose
#----------------------------------------------------------------------
@@ -37,15 +30,11 @@ class DBTxn_distributed(unittest.TestCase):
self.db = db.DB(self.dbenv)
self.db.set_re_len(db.DB_GID_SIZE)
if must_open_db :
- if db.version() >= (4,2) :
- txn=self.dbenv.txn_begin()
- self.db.open(self.filename,
- db.DB_QUEUE, db.DB_CREATE | db.DB_THREAD, 0666,
- txn=txn)
- txn.commit()
- else :
- self.db.open(self.filename,
- db.DB_QUEUE, db.DB_CREATE | db.DB_THREAD, 0666)
+ txn=self.dbenv.txn_begin()
+ self.db.open(self.filename,
+ db.DB_QUEUE, db.DB_CREATE | db.DB_THREAD, 0666,
+ txn=txn)
+ txn.commit()
def setUp(self) :
self.homeDir = get_new_environment_path()
diff --git a/Lib/bsddb/test/test_early_close.py b/Lib/bsddb/test/test_early_close.py
index 86b86dc..e925279 100644
--- a/Lib/bsddb/test/test_early_close.py
+++ b/Lib/bsddb/test/test_early_close.py
@@ -174,7 +174,7 @@ class DBEnvClosedEarlyCrash(unittest.TestCase):
txn.commit()
warnings.resetwarnings()
else :
- # When we drop support for python 2.3 and 2.4
+ # When we drop support for python 2.4
# we could use: (in 2.5 we need a __future__ statement)
#
# with warnings.catch_warnings():
@@ -182,7 +182,7 @@ class DBEnvClosedEarlyCrash(unittest.TestCase):
# txn.commit()
#
# We can not use "with" as is, because it would be invalid syntax
- # in python 2.3, 2.4 and (with no __future__) 2.5.
+ # in python 2.4 and (with no __future__) 2.5.
# Here we simulate "with" following PEP 343 :
w = warnings.catch_warnings()
w.__enter__()
@@ -194,15 +194,14 @@ class DBEnvClosedEarlyCrash(unittest.TestCase):
self.assertRaises(db.DBCursorClosedError, c2.first)
- if db.version() > (4,3,0) :
- def test07_close_db_before_sequence(self):
- import os.path
- path=os.path.join(self.homeDir,self.filename)
- d = db.DB()
- d.open(path, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
- dbs=db.DBSequence(d)
- d.close() # This "close" should close the child DBSequence also
- dbs.close() # If not closed, core dump (in Berkeley DB 4.6.*)
+ def test07_close_db_before_sequence(self):
+ import os.path
+ path=os.path.join(self.homeDir,self.filename)
+ d = db.DB()
+ d.open(path, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
+ dbs=db.DBSequence(d)
+ d.close() # This "close" should close the child DBSequence also
+ dbs.close() # If not closed, core dump (in Berkeley DB 4.6.*)
#----------------------------------------------------------------------
diff --git a/Lib/bsddb/test/test_lock.py b/Lib/bsddb/test/test_lock.py
index 25260fc..10ca8d6 100644
--- a/Lib/bsddb/test/test_lock.py
+++ b/Lib/bsddb/test/test_lock.py
@@ -19,12 +19,6 @@ if have_threads :
#----------------------------------------------------------------------
class LockingTestCase(unittest.TestCase):
- import sys
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
-
def setUp(self):
self.homeDir = get_new_environment_path()
self.env = db.DBEnv()
@@ -89,7 +83,6 @@ class LockingTestCase(unittest.TestCase):
for t in threads:
t.join()
- if db.version() >= (4, 2) :
def test03_lock_timeout(self):
self.env.set_timeout(0, db.DB_SET_LOCK_TIMEOUT)
self.assertEqual(self.env.get_timeout(db.DB_SET_LOCK_TIMEOUT), 0)
diff --git a/Lib/bsddb/test/test_misc.py b/Lib/bsddb/test/test_misc.py
index 8dda296..b1e928f 100644
--- a/Lib/bsddb/test/test_misc.py
+++ b/Lib/bsddb/test/test_misc.py
@@ -9,13 +9,6 @@ from test_all import db, dbshelve, hashopen, test_support, get_new_environment_p
#----------------------------------------------------------------------
class MiscTestCase(unittest.TestCase):
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr, msg=msg)
-
- def assertFalse(self, expr, msg=None):
- self.failIf(expr, msg=msg)
-
def setUp(self):
self.filename = get_new_database_path()
self.homeDir = get_new_environment_path()
@@ -97,10 +90,6 @@ class MiscTestCase(unittest.TestCase):
test_support.unlink(self.filename)
def test07_DB_set_flags_persists(self):
- if db.version() < (4,2):
- # The get_flags API required for this to work is only available
- # in Berkeley DB >= 4.2
- return
try:
db1 = db.DB()
db1.set_flags(db.DB_DUPSORT)
diff --git a/Lib/bsddb/test/test_queue.py b/Lib/bsddb/test/test_queue.py
index 251a8cf..c213ff4 100644
--- a/Lib/bsddb/test/test_queue.py
+++ b/Lib/bsddb/test/test_queue.py
@@ -99,11 +99,6 @@ class SimpleQueueTestCase(unittest.TestCase):
print '\n', '-=' * 30
print "Running %s.test02_basicPost32..." % self.__class__.__name__
- if db.version() < (3, 2, 0):
- if verbose:
- print "Test not run, DB not new enough..."
- return
-
d = db.DB()
d.set_re_len(40) # Queues must be fixed length
d.open(self.filename, db.DB_QUEUE, db.DB_CREATE)
diff --git a/Lib/bsddb/test/test_recno.py b/Lib/bsddb/test/test_recno.py
index 3191fdc..fb6956a 100644
--- a/Lib/bsddb/test/test_recno.py
+++ b/Lib/bsddb/test/test_recno.py
@@ -14,12 +14,6 @@ letters = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
#----------------------------------------------------------------------
class SimpleRecnoTestCase(unittest.TestCase):
- if sys.version_info < (2, 4) :
- def assertFalse(self, expr, msg=None) :
- return self.failIf(expr,msg=msg)
- def assertTrue(self, expr, msg=None) :
- return self.assertTrue(expr, msg=msg)
-
if (sys.version_info < (2, 7)) or ((sys.version_info >= (3, 0)) and
(sys.version_info < (3, 2))) :
def assertIsInstance(self, obj, datatype, msg=None) :
@@ -236,7 +230,9 @@ class SimpleRecnoTestCase(unittest.TestCase):
d.close()
# get the text from the backing source
- text = open(source, 'r').read()
+ f = open(source, 'r')
+ text = f.read()
+ f.close()
text = text.strip()
if verbose:
print text
@@ -256,7 +252,9 @@ class SimpleRecnoTestCase(unittest.TestCase):
d.sync()
d.close()
- text = open(source, 'r').read()
+ f = open(source, 'r')
+ text = f.read()
+ f.close()
text = text.strip()
if verbose:
print text
@@ -298,6 +296,18 @@ class SimpleRecnoTestCase(unittest.TestCase):
c.close()
d.close()
+ def test04_get_size_empty(self) :
+ d = db.DB()
+ d.open(self.filename, dbtype=db.DB_RECNO, flags=db.DB_CREATE)
+
+ row_id = d.append(' ')
+ self.assertEqual(1, d.get_size(key=row_id))
+ row_id = d.append('')
+ self.assertEqual(0, d.get_size(key=row_id))
+
+
+
+
#----------------------------------------------------------------------
diff --git a/Lib/bsddb/test/test_replication.py b/Lib/bsddb/test/test_replication.py
index f3e974f..12ab2dd 100644
--- a/Lib/bsddb/test/test_replication.py
+++ b/Lib/bsddb/test/test_replication.py
@@ -12,11 +12,6 @@ from test_all import db, test_support, have_threads, verbose, \
#----------------------------------------------------------------------
class DBReplication(unittest.TestCase) :
- import sys
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
def setUp(self) :
self.homeDirMaster = get_new_environment_path()
self.homeDirClient = get_new_environment_path()
@@ -76,13 +71,57 @@ class DBReplication(unittest.TestCase) :
class DBReplicationManager(DBReplication) :
def test01_basic_replication(self) :
master_port = test_support.find_unused_port()
- self.dbenvMaster.repmgr_set_local_site("127.0.0.1", master_port)
client_port = test_support.find_unused_port()
- self.dbenvClient.repmgr_set_local_site("127.0.0.1", client_port)
- self.dbenvMaster.repmgr_add_remote_site("127.0.0.1", client_port)
- self.dbenvClient.repmgr_add_remote_site("127.0.0.1", master_port)
- self.dbenvMaster.rep_set_nsites(2)
- self.dbenvClient.rep_set_nsites(2)
+ if db.version() >= (5, 2) :
+ self.site = self.dbenvMaster.repmgr_site("127.0.0.1", master_port)
+ self.site.set_config(db.DB_GROUP_CREATOR, True)
+ self.site.set_config(db.DB_LOCAL_SITE, True)
+ self.site2 = self.dbenvMaster.repmgr_site("127.0.0.1", client_port)
+
+ self.site3 = self.dbenvClient.repmgr_site("127.0.0.1", master_port)
+ self.site3.set_config(db.DB_BOOTSTRAP_HELPER, True)
+ self.site4 = self.dbenvClient.repmgr_site("127.0.0.1", client_port)
+ self.site4.set_config(db.DB_LOCAL_SITE, True)
+
+ d = {
+ db.DB_BOOTSTRAP_HELPER: [False, False, True, False],
+ db.DB_GROUP_CREATOR: [True, False, False, False],
+ db.DB_LEGACY: [False, False, False, False],
+ db.DB_LOCAL_SITE: [True, False, False, True],
+ db.DB_REPMGR_PEER: [False, False, False, False ],
+ }
+
+ for i, j in d.items() :
+ for k, v in \
+ zip([self.site, self.site2, self.site3, self.site4], j) :
+ if v :
+ self.assertTrue(k.get_config(i))
+ else :
+ self.assertFalse(k.get_config(i))
+
+ self.assertNotEqual(self.site.get_eid(), self.site2.get_eid())
+ self.assertNotEqual(self.site3.get_eid(), self.site4.get_eid())
+
+ for i, j in zip([self.site, self.site2, self.site3, self.site4], \
+ [master_port, client_port, master_port, client_port]) :
+ addr = i.get_address()
+ self.assertEqual(addr, ("127.0.0.1", j))
+
+ for i in [self.site, self.site2] :
+ self.assertEqual(i.get_address(),
+ self.dbenvMaster.repmgr_site_by_eid(i.get_eid()).get_address())
+ for i in [self.site3, self.site4] :
+ self.assertEqual(i.get_address(),
+ self.dbenvClient.repmgr_site_by_eid(i.get_eid()).get_address())
+ else :
+ self.dbenvMaster.repmgr_set_local_site("127.0.0.1", master_port)
+ self.dbenvClient.repmgr_set_local_site("127.0.0.1", client_port)
+ self.dbenvMaster.repmgr_add_remote_site("127.0.0.1", client_port)
+ self.dbenvClient.repmgr_add_remote_site("127.0.0.1", master_port)
+
+ self.dbenvMaster.rep_set_nsites(2)
+ self.dbenvClient.rep_set_nsites(2)
+
self.dbenvMaster.rep_set_priority(10)
self.dbenvClient.rep_set_priority(0)
@@ -144,17 +183,19 @@ class DBReplicationManager(DBReplication) :
d = self.dbenvMaster.repmgr_site_list()
self.assertEqual(len(d), 1)
- self.assertEqual(d[0][0], "127.0.0.1")
- self.assertEqual(d[0][1], client_port)
- self.assertTrue((d[0][2]==db.DB_REPMGR_CONNECTED) or \
- (d[0][2]==db.DB_REPMGR_DISCONNECTED))
+ d = d.values()[0] # There is only one
+ self.assertEqual(d[0], "127.0.0.1")
+ self.assertEqual(d[1], client_port)
+ self.assertTrue((d[2]==db.DB_REPMGR_CONNECTED) or \
+ (d[2]==db.DB_REPMGR_DISCONNECTED))
d = self.dbenvClient.repmgr_site_list()
self.assertEqual(len(d), 1)
- self.assertEqual(d[0][0], "127.0.0.1")
- self.assertEqual(d[0][1], master_port)
- self.assertTrue((d[0][2]==db.DB_REPMGR_CONNECTED) or \
- (d[0][2]==db.DB_REPMGR_DISCONNECTED))
+ d = d.values()[0] # There is only one
+ self.assertEqual(d[0], "127.0.0.1")
+ self.assertEqual(d[1], master_port)
+ self.assertTrue((d[2]==db.DB_REPMGR_CONNECTED) or \
+ (d[2]==db.DB_REPMGR_DISCONNECTED))
if db.version() >= (4,6) :
d = self.dbenvMaster.repmgr_stat(flags=db.DB_STAT_CLEAR);
@@ -461,6 +502,13 @@ class DBBaseReplication(DBReplication) :
self.assertTrue(self.confirmed_master)
+ # Race condition showed up after upgrading to Solaris 10 Update 10
+ # https://forums.oracle.com/forums/thread.jspa?messageID=9902860
+ # jcea@jcea.es: See private email from Paula Bingham (Oracle),
+ # in 20110929.
+ while not (self.dbenvClient.rep_stat()["startup_complete"]) :
+ pass
+
if db.version() >= (4,7) :
def test04_test_clockskew(self) :
fast, slow = 1234, 1230
diff --git a/Lib/bsddb/test/test_sequence.py b/Lib/bsddb/test/test_sequence.py
index 3d3ee29..f0aa12a 100644
--- a/Lib/bsddb/test/test_sequence.py
+++ b/Lib/bsddb/test/test_sequence.py
@@ -5,11 +5,6 @@ from test_all import db, test_support, get_new_environment_path, get_new_databas
class DBSequenceTest(unittest.TestCase):
- import sys
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
def setUp(self):
self.int_32_max = 0x100000000
self.homeDir = get_new_environment_path()
@@ -133,8 +128,7 @@ class DBSequenceTest(unittest.TestCase):
def test_suite():
suite = unittest.TestSuite()
- if db.version() >= (4,3):
- suite.addTest(unittest.makeSuite(DBSequenceTest))
+ suite.addTest(unittest.makeSuite(DBSequenceTest))
return suite
diff --git a/Lib/bsddb/test/test_thread.py b/Lib/bsddb/test/test_thread.py
index 91002b8..42212e9 100644
--- a/Lib/bsddb/test/test_thread.py
+++ b/Lib/bsddb/test/test_thread.py
@@ -35,10 +35,6 @@ class BaseThreadedTestCase(unittest.TestCase):
dbsetflags = 0
envflags = 0
- if sys.version_info < (2, 4) :
- def assertTrue(self, expr, msg=None):
- self.failUnless(expr,msg=msg)
-
def setUp(self):
if verbose:
dbutils._deadlock_VerboseFile = sys.stdout
generated by cgit v1.2.3 (git 2.25.1) at 2024-07-04 17:36:33 +0000