526 lines
16 KiB
Python
526 lines
16 KiB
Python
"""Run all test cases.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import unittest
|
|
try:
|
|
# For Pythons w/distutils pybsddb
|
|
import bsddb3 as bsddb
|
|
except ImportError:
|
|
# For Python 2.3
|
|
import bsddb
|
|
|
|
|
|
if sys.version_info[0] >= 3 :
|
|
charset = "iso8859-1" # Full 8 bit
|
|
|
|
class cursor_py3k(object) :
|
|
def __init__(self, db, *args, **kwargs) :
|
|
self._dbcursor = db.cursor(*args, **kwargs)
|
|
|
|
def __getattr__(self, v) :
|
|
return getattr(self._dbcursor, v)
|
|
|
|
def _fix(self, v) :
|
|
if v == None : return None
|
|
key, value = v
|
|
if isinstance(key, bytes) :
|
|
key = key.decode(charset)
|
|
return (key, value.decode(charset))
|
|
|
|
def __next__(self) :
|
|
v = getattr(self._dbcursor, "next")()
|
|
return self._fix(v)
|
|
|
|
next = __next__
|
|
|
|
def previous(self) :
|
|
v = self._dbcursor.previous()
|
|
return self._fix(v)
|
|
|
|
def last(self) :
|
|
v = self._dbcursor.last()
|
|
return self._fix(v)
|
|
|
|
def set(self, k) :
|
|
if isinstance(k, str) :
|
|
k = bytes(k, charset)
|
|
v = self._dbcursor.set(k)
|
|
return self._fix(v)
|
|
|
|
def set_recno(self, num) :
|
|
v = self._dbcursor.set_recno(num)
|
|
return self._fix(v)
|
|
|
|
def set_range(self, k, dlen=-1, doff=-1) :
|
|
if isinstance(k, str) :
|
|
k = bytes(k, charset)
|
|
v = self._dbcursor.set_range(k, dlen=dlen, doff=doff)
|
|
return self._fix(v)
|
|
|
|
def dup(self, flags=0) :
|
|
cursor = self._dbcursor.dup(flags)
|
|
return dup_cursor_py3k(cursor)
|
|
|
|
def next_dup(self) :
|
|
v = self._dbcursor.next_dup()
|
|
return self._fix(v)
|
|
|
|
def next_nodup(self) :
|
|
v = self._dbcursor.next_nodup()
|
|
return self._fix(v)
|
|
|
|
def put(self, key, value, flags=0, dlen=-1, doff=-1) :
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
if isinstance(value, str) :
|
|
value = bytes(value, charset)
|
|
return self._dbcursor.put(key, value, flags=flags, dlen=dlen,
|
|
doff=doff)
|
|
|
|
def current(self, flags=0, dlen=-1, doff=-1) :
|
|
v = self._dbcursor.current(flags=flags, dlen=dlen, doff=doff)
|
|
return self._fix(v)
|
|
|
|
def first(self) :
|
|
v = self._dbcursor.first()
|
|
return self._fix(v)
|
|
|
|
def pget(self, key=None, data=None, flags=0) :
|
|
# Incorrect because key can be a bare number,
|
|
# but enough to pass testsuite
|
|
if isinstance(key, int) and (data==None) and (flags==0) :
|
|
flags = key
|
|
key = None
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
if isinstance(data, int) and (flags==0) :
|
|
flags = data
|
|
data = None
|
|
if isinstance(data, str) :
|
|
data = bytes(data, charset)
|
|
v=self._dbcursor.pget(key=key, data=data, flags=flags)
|
|
if v != None :
|
|
v1, v2, v3 = v
|
|
if isinstance(v1, bytes) :
|
|
v1 = v1.decode(charset)
|
|
if isinstance(v2, bytes) :
|
|
v2 = v2.decode(charset)
|
|
|
|
v = (v1, v2, v3.decode(charset))
|
|
|
|
return v
|
|
|
|
def join_item(self) :
|
|
v = self._dbcursor.join_item()
|
|
if v != None :
|
|
v = v.decode(charset)
|
|
return v
|
|
|
|
def get(self, *args, **kwargs) :
|
|
l = len(args)
|
|
if l == 2 :
|
|
k, f = args
|
|
if isinstance(k, str) :
|
|
k = bytes(k, "iso8859-1")
|
|
args = (k, f)
|
|
elif l == 3 :
|
|
k, d, f = args
|
|
if isinstance(k, str) :
|
|
k = bytes(k, charset)
|
|
if isinstance(d, str) :
|
|
d = bytes(d, charset)
|
|
args =(k, d, f)
|
|
|
|
v = self._dbcursor.get(*args, **kwargs)
|
|
if v != None :
|
|
k, v = v
|
|
if isinstance(k, bytes) :
|
|
k = k.decode(charset)
|
|
v = (k, v.decode(charset))
|
|
return v
|
|
|
|
def get_both(self, key, value) :
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
if isinstance(value, str) :
|
|
value = bytes(value, charset)
|
|
v=self._dbcursor.get_both(key, value)
|
|
return self._fix(v)
|
|
|
|
class dup_cursor_py3k(cursor_py3k) :
|
|
def __init__(self, dbcursor) :
|
|
self._dbcursor = dbcursor
|
|
|
|
class DB_py3k(object) :
|
|
def __init__(self, *args, **kwargs) :
|
|
args2=[]
|
|
for i in args :
|
|
if isinstance(i, DBEnv_py3k) :
|
|
i = i._dbenv
|
|
args2.append(i)
|
|
args = tuple(args2)
|
|
for k, v in kwargs.items() :
|
|
if isinstance(v, DBEnv_py3k) :
|
|
kwargs[k] = v._dbenv
|
|
|
|
self._db = bsddb._db.DB_orig(*args, **kwargs)
|
|
|
|
def __contains__(self, k) :
|
|
if isinstance(k, str) :
|
|
k = bytes(k, charset)
|
|
return getattr(self._db, "has_key")(k)
|
|
|
|
def __getitem__(self, k) :
|
|
if isinstance(k, str) :
|
|
k = bytes(k, charset)
|
|
v = self._db[k]
|
|
if v != None :
|
|
v = v.decode(charset)
|
|
return v
|
|
|
|
def __setitem__(self, k, v) :
|
|
if isinstance(k, str) :
|
|
k = bytes(k, charset)
|
|
if isinstance(v, str) :
|
|
v = bytes(v, charset)
|
|
self._db[k] = v
|
|
|
|
def __delitem__(self, k) :
|
|
if isinstance(k, str) :
|
|
k = bytes(k, charset)
|
|
del self._db[k]
|
|
|
|
def __getattr__(self, v) :
|
|
return getattr(self._db, v)
|
|
|
|
def __len__(self) :
|
|
return len(self._db)
|
|
|
|
def has_key(self, k, txn=None) :
|
|
if isinstance(k, str) :
|
|
k = bytes(k, charset)
|
|
return self._db.has_key(k, txn=txn)
|
|
|
|
def put(self, key, value, txn=None, flags=0, dlen=-1, doff=-1) :
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
if isinstance(value, str) :
|
|
value = bytes(value, charset)
|
|
return self._db.put(key, value, flags=flags, txn=txn, dlen=dlen,
|
|
doff=doff)
|
|
|
|
def append(self, value, txn=None) :
|
|
if isinstance(value, str) :
|
|
value = bytes(value, charset)
|
|
return self._db.append(value, txn=txn)
|
|
|
|
def get_size(self, key) :
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
return self._db.get_size(key)
|
|
|
|
def get(self, key, default="MagicCookie", txn=None, flags=0, dlen=-1, doff=-1) :
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
if default != "MagicCookie" : # Magic for 'test_get_none.py'
|
|
v=self._db.get(key, default=default, txn=txn, flags=flags,
|
|
dlen=dlen, doff=doff)
|
|
else :
|
|
v=self._db.get(key, txn=txn, flags=flags,
|
|
dlen=dlen, doff=doff)
|
|
if (v != None) and isinstance(v, bytes) :
|
|
v = v.decode(charset)
|
|
return v
|
|
|
|
def pget(self, key, txn=None) :
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
v=self._db.pget(key, txn=txn)
|
|
if v != None :
|
|
v1, v2 = v
|
|
if isinstance(v1, bytes) :
|
|
v1 = v1.decode(charset)
|
|
|
|
v = (v1, v2.decode(charset))
|
|
return v
|
|
|
|
def get_both(self, key, value, txn=None, flags=0) :
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
if isinstance(value, str) :
|
|
value = bytes(value, charset)
|
|
v=self._db.get_both(key, value, txn=txn, flags=flags)
|
|
if v != None :
|
|
v = v.decode(charset)
|
|
return v
|
|
|
|
def delete(self, key, txn=None) :
|
|
if isinstance(key, str) :
|
|
key = bytes(key, charset)
|
|
return self._db.delete(key, txn=txn)
|
|
|
|
def keys(self) :
|
|
k = self._db.keys()
|
|
if len(k) and isinstance(k[0], bytes) :
|
|
return [i.decode(charset) for i in self._db.keys()]
|
|
else :
|
|
return k
|
|
|
|
def items(self) :
|
|
data = self._db.items()
|
|
if not len(data) : return data
|
|
data2 = []
|
|
for k, v in data :
|
|
if isinstance(k, bytes) :
|
|
k = k.decode(charset)
|
|
data2.append((k, v.decode(charset)))
|
|
return data2
|
|
|
|
def associate(self, secondarydb, callback, flags=0, txn=None) :
|
|
class associate_callback(object) :
|
|
def __init__(self, callback) :
|
|
self._callback = callback
|
|
|
|
def callback(self, key, data) :
|
|
if isinstance(key, str) :
|
|
key = key.decode(charset)
|
|
data = data.decode(charset)
|
|
key = self._callback(key, data)
|
|
if (key != bsddb._db.DB_DONOTINDEX) and isinstance(key,
|
|
str) :
|
|
key = bytes(key, charset)
|
|
return key
|
|
|
|
return self._db.associate(secondarydb._db,
|
|
associate_callback(callback).callback, flags=flags, txn=txn)
|
|
|
|
def cursor(self, txn=None, flags=0) :
|
|
return cursor_py3k(self._db, txn=txn, flags=flags)
|
|
|
|
def join(self, cursor_list) :
|
|
cursor_list = [i._dbcursor for i in cursor_list]
|
|
return dup_cursor_py3k(self._db.join(cursor_list))
|
|
|
|
class DBEnv_py3k(object) :
|
|
def __init__(self, *args, **kwargs) :
|
|
self._dbenv = bsddb._db.DBEnv_orig(*args, **kwargs)
|
|
|
|
def __getattr__(self, v) :
|
|
return getattr(self._dbenv, v)
|
|
|
|
class DBSequence_py3k(object) :
|
|
def __init__(self, db, *args, **kwargs) :
|
|
self._db=db
|
|
self._dbsequence = bsddb._db.DBSequence_orig(db._db, *args, **kwargs)
|
|
|
|
def __getattr__(self, v) :
|
|
return getattr(self._dbsequence, v)
|
|
|
|
def open(self, key, *args, **kwargs) :
|
|
return self._dbsequence.open(bytes(key, charset), *args, **kwargs)
|
|
|
|
def get_key(self) :
|
|
return self._dbsequence.get_key().decode(charset)
|
|
|
|
def get_dbp(self) :
|
|
return self._db
|
|
|
|
import string
|
|
string.letters=[chr(i) for i in xrange(65,91)]
|
|
|
|
bsddb._db.DBEnv_orig = bsddb._db.DBEnv
|
|
bsddb._db.DB_orig = bsddb._db.DB
|
|
bsddb._db.DBSequence_orig = bsddb._db.DBSequence
|
|
|
|
def do_proxy_db_py3k(flag) :
|
|
flag2 = do_proxy_db_py3k.flag
|
|
do_proxy_db_py3k.flag = flag
|
|
if flag :
|
|
bsddb.DBEnv = bsddb.db.DBEnv = bsddb._db.DBEnv = DBEnv_py3k
|
|
bsddb.DB = bsddb.db.DB = bsddb._db.DB = DB_py3k
|
|
bsddb._db.DBSequence = DBSequence_py3k
|
|
else :
|
|
bsddb.DBEnv = bsddb.db.DBEnv = bsddb._db.DBEnv = bsddb._db.DBEnv_orig
|
|
bsddb.DB = bsddb.db.DB = bsddb._db.DB = bsddb._db.DB_orig
|
|
bsddb._db.DBSequence = bsddb._db.DBSequence_orig
|
|
return flag2
|
|
|
|
do_proxy_db_py3k.flag = False
|
|
do_proxy_db_py3k(True)
|
|
|
|
try:
|
|
# For Pythons w/distutils pybsddb
|
|
from bsddb3 import db, dbtables, dbutils, dbshelve, \
|
|
hashopen, btopen, rnopen, dbobj
|
|
except ImportError:
|
|
# For Python 2.3
|
|
from bsddb import db, dbtables, dbutils, dbshelve, \
|
|
hashopen, btopen, rnopen, dbobj
|
|
|
|
try:
|
|
from bsddb3 import test_support
|
|
except ImportError:
|
|
if sys.version_info[0] < 3 :
|
|
from test import test_support
|
|
else :
|
|
from test import support as test_support
|
|
|
|
|
|
try:
|
|
if sys.version_info[0] < 3 :
|
|
from threading import Thread, currentThread
|
|
del Thread, currentThread
|
|
else :
|
|
from threading import Thread, current_thread
|
|
del Thread, current_thread
|
|
have_threads = True
|
|
except ImportError:
|
|
have_threads = False
|
|
|
|
verbose = 0
|
|
if 'verbose' in sys.argv:
|
|
verbose = 1
|
|
sys.argv.remove('verbose')
|
|
|
|
if 'silent' in sys.argv: # take care of old flag, just in case
|
|
verbose = 0
|
|
sys.argv.remove('silent')
|
|
|
|
|
|
def print_versions():
|
|
print
|
|
print '-=' * 38
|
|
print db.DB_VERSION_STRING
|
|
print 'bsddb.db.version(): %s' % (db.version(), )
|
|
print 'bsddb.db.__version__: %s' % db.__version__
|
|
print 'bsddb.db.cvsid: %s' % db.cvsid
|
|
print 'py module: %s' % bsddb.__file__
|
|
print 'extension module: %s' % bsddb._bsddb.__file__
|
|
print 'python version: %s' % sys.version
|
|
print 'My pid: %s' % os.getpid()
|
|
print '-=' * 38
|
|
|
|
|
|
def get_new_path(name) :
|
|
get_new_path.mutex.acquire()
|
|
try :
|
|
import os
|
|
path=os.path.join(get_new_path.prefix,
|
|
name+"_"+str(os.getpid())+"_"+str(get_new_path.num))
|
|
get_new_path.num+=1
|
|
finally :
|
|
get_new_path.mutex.release()
|
|
return path
|
|
|
|
def get_new_environment_path() :
|
|
path=get_new_path("environment")
|
|
import os
|
|
try:
|
|
os.makedirs(path,mode=0700)
|
|
except os.error:
|
|
test_support.rmtree(path)
|
|
os.makedirs(path)
|
|
return path
|
|
|
|
def get_new_database_path() :
|
|
path=get_new_path("database")
|
|
import os
|
|
if os.path.exists(path) :
|
|
os.remove(path)
|
|
return 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.num=0
|
|
|
|
def get_test_path_prefix() :
|
|
return get_new_path.prefix
|
|
|
|
def set_test_path_prefix(path) :
|
|
get_new_path.prefix=path
|
|
|
|
def remove_test_path_directory() :
|
|
test_support.rmtree(get_new_path.prefix)
|
|
|
|
if have_threads :
|
|
import threading
|
|
get_new_path.mutex=threading.Lock()
|
|
del threading
|
|
else :
|
|
class Lock(object) :
|
|
def acquire(self) :
|
|
pass
|
|
def release(self) :
|
|
pass
|
|
get_new_path.mutex=Lock()
|
|
del Lock
|
|
|
|
|
|
|
|
class PrintInfoFakeTest(unittest.TestCase):
|
|
def testPrintVersions(self):
|
|
print_versions()
|
|
|
|
|
|
# This little hack is for when this module is run as main and all the
|
|
# other modules import it so they will still be able to get the right
|
|
# verbose setting. It's confusing but it works.
|
|
if sys.version_info[0] < 3 :
|
|
import test_all
|
|
test_all.verbose = verbose
|
|
else :
|
|
import sys
|
|
print >>sys.stderr, "Work to do!"
|
|
|
|
|
|
def suite(module_prefix='', timing_check=None):
|
|
test_modules = [
|
|
'test_associate',
|
|
'test_basics',
|
|
'test_compare',
|
|
'test_compat',
|
|
'test_cursor_pget_bug',
|
|
'test_dbobj',
|
|
'test_dbshelve',
|
|
'test_dbtables',
|
|
'test_distributed_transactions',
|
|
'test_early_close',
|
|
'test_get_none',
|
|
'test_join',
|
|
'test_lock',
|
|
'test_misc',
|
|
'test_pickle',
|
|
'test_queue',
|
|
'test_recno',
|
|
'test_replication',
|
|
'test_sequence',
|
|
'test_thread',
|
|
]
|
|
|
|
alltests = unittest.TestSuite()
|
|
for name in test_modules:
|
|
#module = __import__(name)
|
|
# Do it this way so that suite may be called externally via
|
|
# python's Lib/test/test_bsddb3.
|
|
module = __import__(module_prefix+name, globals(), locals(), name)
|
|
|
|
alltests.addTest(module.test_suite())
|
|
if timing_check:
|
|
alltests.addTest(unittest.makeSuite(timing_check))
|
|
return alltests
|
|
|
|
|
|
def test_suite():
|
|
suite = unittest.TestSuite()
|
|
suite.addTest(unittest.makeSuite(PrintInfoFakeTest))
|
|
return suite
|
|
|
|
|
|
if __name__ == '__main__':
|
|
print_versions()
|
|
unittest.main(defaultTest='suite')
|