448 lines
15 KiB
Python
448 lines
15 KiB
Python
"""
|
|
TestCases for python DB duplicate and Btree key comparison function.
|
|
"""
|
|
|
|
import sys, os, re
|
|
import test_all
|
|
from cStringIO import StringIO
|
|
|
|
import unittest
|
|
|
|
from test_all import db, dbshelve, test_support, \
|
|
get_new_environment_path, get_new_database_path
|
|
|
|
|
|
# Needed for python 3. "cmp" vanished in 3.0.1
|
|
def cmp(a, b) :
|
|
if a==b : return 0
|
|
if a<b : return -1
|
|
return 1
|
|
|
|
lexical_cmp = cmp
|
|
|
|
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)
|
|
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) :
|
|
def comparator_test_helper(self, comparator, expected_data) :
|
|
data = expected_data[:]
|
|
|
|
import sys
|
|
if sys.version_info < (2, 6) :
|
|
data.sort(cmp=comparator)
|
|
else : # Insertion Sort. Please, improve
|
|
data2 = []
|
|
for i in data :
|
|
for j, k in enumerate(data2) :
|
|
r = comparator(k, i)
|
|
if r == 1 :
|
|
data2.insert(j, i)
|
|
break
|
|
else :
|
|
data2.append(i)
|
|
data = data2
|
|
|
|
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) :
|
|
rev = _expected_lexical_test_data[:]
|
|
rev.reverse()
|
|
self.comparator_test_helper(make_reverse_comparator(lexical_cmp),
|
|
rev)
|
|
def test_lowercase_comparator(self) :
|
|
self.comparator_test_helper(lowercase_cmp,
|
|
_expected_lowercase_test_data)
|
|
|
|
class AbstractBtreeKeyCompareTestCase(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) :
|
|
i = 0
|
|
for item in data:
|
|
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 setupDB(self, key_comparator) :
|
|
self.db.set_bt_compare(key_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:
|
|
key, ignore = rec
|
|
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),
|
|
"not enough values returned from cursor")
|
|
finally:
|
|
curs.close()
|
|
|
|
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_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)
|
|
|
|
def test_compare_function_useless(self) :
|
|
self.startTest()
|
|
def socialist_comparator(l, r) :
|
|
return 0
|
|
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'])
|
|
|
|
|
|
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 check_results(self, results) :
|
|
pass
|
|
|
|
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()
|
|
|
|
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_bt_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_bt_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()
|
|
|
|
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()
|
|
|
|
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_dup_compare, my_compare)
|
|
|
|
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(DuplicateExceptionsTestCase))
|
|
res.addTest(unittest.makeSuite(DuplicateCompareTestCase))
|
|
return res
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(defaultTest = 'suite')
|