diff options
Diffstat (limited to 'Lib/bsddb/test/test_all.py')
| -rw-r--r-- | Lib/bsddb/test/test_all.py | 465 |
1 files changed, 443 insertions, 22 deletions
diff --git a/Lib/bsddb/test/test_all.py b/Lib/bsddb/test/test_all.py index b818a33..a4ddd66 100644 --- a/Lib/bsddb/test/test_all.py +++ b/Lib/bsddb/test/test_all.py @@ -6,18 +6,377 @@ import os import unittest try: # For Pythons w/distutils pybsddb - from bsddb3 import db + import bsddb3 as bsddb except ImportError: # For Python 2.3 - from bsddb import db + import bsddb -verbose = False + +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) + + 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 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 list(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 = list(self._db.keys()) + if len(k) and isinstance(k[0], bytes) : + return [i.decode(charset) for i in list(self._db.keys())] + else : + return k + + def items(self) : + data = list(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 range(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: + from test import 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 = True + verbose = 1 sys.argv.remove('verbose') if 'silent' in sys.argv: # take care of old flag, just in case - verbose = False + verbose = 0 sys.argv.remove('silent') @@ -28,11 +387,71 @@ def print_versions(): 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=0o700) + 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() @@ -41,30 +460,26 @@ class PrintInfoFakeTest(unittest.TestCase): # 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. -try: - import test_all -except ImportError: - pass -else: +if sys.version_info[0] < 3 : + from . import test_all test_all.verbose = verbose +else : + import sys + print("Work to do!", file=sys.stderr) -def suite(): - try: - # this is special, it used to segfault the interpreter - import test_1413192 - except: - pass - +def suite(module_prefix='', timing_check=None): test_modules = [ 'test_associate', 'test_basics', - 'test_compat', 'test_compare', + 'test_compat', + 'test_cursor_pget_bug', 'test_dbobj', 'test_dbshelve', 'test_dbtables', - 'test_env_close', + 'test_distributed_transactions', + 'test_early_close', 'test_get_none', 'test_join', 'test_lock', @@ -72,15 +487,21 @@ def suite(): 'test_pickle', 'test_queue', 'test_recno', - 'test_thread', + 'test_replication', 'test_sequence', - 'test_cursor_pget_bug', + 'test_thread', ] alltests = unittest.TestSuite() for name in test_modules: - module = __import__(name) + #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 |