# software has been tested, but no warranty is expressed or
# implied.
#
-# -- Gregory P. Smith <greg@electricrain.com>
+# -- Gregory P. Smith <greg@krypto.org>
# This provides a simple database table interface built on top of
# the Python BerkeleyDB 3 interface.
# software has been tested, but no warranty is expressed or
# implied.
#
-# Author: Gregory P. Smith <greg@electricrain.com>
+# Author: Gregory P. Smith <greg@krypto.org>
#
# Note: I don't know how useful this is in reality since when a
# DBLockDeadlockError happens the current transaction is supposed to be
#
#------------------------------------------------------------------------
-
-#
-# import the time.sleep function in a namespace safe way to allow
-# "from bsddb.dbutils import *"
-#
-from time import sleep as _sleep
-
+import time
from . import db
# always sleep at least N seconds between retrys
while True:
try:
return function(*_args, **_kwargs)
- except db.DBLockDeadlockError:
+ except db.DBLockDeadlockError as e:
if _deadlock_VerboseFile:
_deadlock_VerboseFile.write(
- 'dbutils.DeadlockWrap: sleeping %1.3f\n' % sleeptime)
- _sleep(sleeptime)
+ 'bsddb.dbutils.DeadlockWrap: ' +
+ 'sleeping %1.3f\n' % sleeptime)
+ time.sleep(sleeptime)
# exponential backoff in the sleep time
sleeptime *= 2
if sleeptime > _deadlock_MaxSleepTime:
sleeptime = _deadlock_MaxSleepTime
max_retries -= 1
if max_retries == -1:
+ if _deadlock_VerboseFile:
+ _deadlock_VerboseFile.write(
+ 'bsddb.dbutils.DeadlockWrap: ' +
+ 'max_retries reached, reraising %s\n' % e)
raise
# For Python 2.3
from bsddb import db
-verbose = 0
+verbose = False
if 'verbose' in sys.argv:
- verbose = 1
+ verbose = True
sys.argv.remove('verbose')
if 'silent' in sys.argv: # take care of old flag, just in case
- verbose = 0
+ verbose = False
sys.argv.remove('silent')
TestCases for DB.associate.
"""
+import shutil
import sys, os
import tempfile
import time
have_threads = 0
import unittest
-from .test_all import verbose
+from bsddb.test.test_all import verbose
try:
# For Pythons w/distutils pybsddb
class AssociateErrorTestCase(unittest.TestCase):
def setUp(self):
self.filename = self.__class__.__name__ + '.db'
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- self.homeDir = homeDir
- try:
- os.mkdir(homeDir)
- except os.error:
- import glob
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ self.homeDir = tempfile.mkdtemp()
self.env = db.DBEnv()
- self.env.open(homeDir, db.DB_CREATE | db.DB_INIT_MPOOL)
+ self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL)
def tearDown(self):
self.env.close()
self.env = None
- import glob
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def test00_associateDBError(self):
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:
- import glob
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ self.homeDir = tempfile.mkdtemp()
self.env = db.DBEnv()
- self.env.open(homeDir, db.DB_CREATE | db.DB_INIT_MPOOL |
+ self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL |
db.DB_INIT_LOCK | db.DB_THREAD | self.envFlags)
def tearDown(self):
self.closeDB()
self.env.close()
self.env = None
- import glob
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def addDataToDB(self, d, txn=None):
for key, value in musicdata.items():
# For Python 2.3
from bsddb import db
-from .test_all import verbose
+from bsddb.test.test_all import verbose
DASH = b'-'
letters = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
def setUp(self):
if self.useEnv:
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- self.homeDir = homeDir
- try:
- shutil.rmtree(homeDir)
- except OSError as e:
- # unix returns ENOENT, windows returns ESRCH
- if e.errno not in (errno.ENOENT, errno.ESRCH): raise
- os.mkdir(homeDir)
+ self.homeDir = tempfile.mkdtemp()
try:
self.env = db.DBEnv()
self.env.set_lg_max(1024*1024)
self.env.set_tx_max(30)
self.env.set_tx_timestamp(int(time.time()))
self.env.set_flags(self.envsetflags, 1)
- self.env.open(homeDir, self.envflags | db.DB_CREATE)
- tempfile.tempdir = homeDir
+ self.env.open(self.homeDir, self.envflags | db.DB_CREATE)
+ old_tempfile_tempdir = tempfile.tempdir
+ tempfile.tempdir = self.homeDir
self.filename = os.path.split(tempfile.mktemp())[1]
- tempfile.tempdir = None
+ tempfile.tempdir = old_tempfile_tempdir
# Yes, a bare except is intended, since we're re-raising the exc.
except:
- shutil.rmtree(homeDir)
+ shutil.rmtree(self.homeDir)
raise
else:
self.env = None
TestCases for python DB Btree key comparison function.
"""
+import shutil
import sys, os, re
from io import StringIO
-
-from . import test_all
+import tempfile
import unittest
try:
lexical_cmp = cmp
def lowercase_cmp(left, right):
- return cmp (left.lower(), right.lower())
+ return cmp (str(left, encoding='ascii').lower(),
+ str(right, encoding='ascii').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']
+_expected_lexical_test_data = [bytes(_) for _ in
+ ('', 'CCCP', 'a', 'aaa', 'b', 'c', 'cccce', 'ccccf')]
+_expected_lowercase_test_data = [bytes(_) for _ in
+ ('', 'a', 'aaa', 'b', 'c', 'CC', 'cccce', 'ccccf', 'CCCP')]
class ComparatorTests (unittest.TestCase):
def comparator_test_helper (self, comparator, expected_data):
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.homeDir = tempfile.mkdtemp()
env = db.DBEnv ()
- env.open (homeDir,
+ env.open (self.homeDir,
db.DB_CREATE | db.DB_INIT_MPOOL
| db.DB_INIT_LOCK | db.DB_THREAD)
self.env = env
if self.env is not None:
self.env.close ()
self.env = None
- import glob
- map (os.remove, glob.glob (os.path.join (self.homeDir, '*')))
+ shutil.rmtree(self.homeDir)
def addDataToDB (self, data):
i = 0
self.failUnless (index < len (expected),
"to many values returned from cursor")
self.failUnless (expected[index] == key,
- "expected value `%s' at %d but got `%s'"
+ "expected value %r at %d but got %r"
% (expected[index], index, key))
index = index + 1
rec = curs.next ()
def socialist_comparator (l, r):
return 0
self.createDB (socialist_comparator)
- self.addDataToDB (['b', 'a', 'd'])
+ self.addDataToDB ([b'b', b'a', b'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'b'])
class BtreeExceptionsTestCase (AbstractBtreeKeyCompareTestCase):
return res
if __name__ == '__main__':
- unittest.main (defaultTest = 'suite')
+ unittest.main (defaultTest = 'test_suite')
import unittest
import tempfile
-from .test_all import verbose
+from bsddb.test.test_all import verbose
from bsddb import db, hashopen, btopen, rnopen
import unittest
import sys, os, glob
+import shutil
+import tempfile
from bsddb import db
db_name = 'test-cursor_pget.db'
def setUp(self):
- self.homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home')
- try:
- os.mkdir(self.homeDir)
- except os.error:
- pass
+ self.homeDir = tempfile.mkdtemp()
self.env = db.DBEnv()
self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL)
self.primary_db = db.DB(self.env)
del self.secondary_db
del self.primary_db
del self.env
- for file in glob.glob(os.path.join(self.homeDir, '*')):
- os.remove(file)
- os.removedirs(self.homeDir)
+ shutil.rmtree(self.homeDir)
def test_pget(self):
cursor = self.secondary_db.cursor()
+import shutil
import sys, os
import unittest
import glob
class dbobjTestCase(unittest.TestCase):
"""Verify that dbobj.DB and dbobj.DBEnv work properly"""
- db_home = 'db_home'
db_name = 'test-dbobj.db'
def setUp(self):
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- self.homeDir = homeDir
- try: os.mkdir(homeDir)
- except os.error: pass
+ self.homeDir = tempfile.mkdtemp()
def tearDown(self):
if hasattr(self, 'db'):
del self.db
if hasattr(self, 'env'):
del self.env
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def test01_both(self):
class TestDBEnv(dbobj.DBEnv): pass
"""
import os
+import shutil
import tempfile, random
import unittest
from bsddb import db, dbshelve
-from .test_all import verbose
+from bsddb.test.test_all import verbose
#----------------------------------------------------------------------
print("Running %s.test01_basics..." % self.__class__.__name__)
self.populateDB(self.d)
+ if verbose:
+ print(1, self.d.keys())
self.d.sync()
+ if verbose:
+ print(2, self.d.keys())
self.do_close()
self.do_open()
+ if verbose:
+ print(3, self.d.keys())
d = self.d
l = len(d)
print("keys:", k)
print("stats:", s)
- assert 0 == d.has_key(b'bad key')
- assert 1 == d.has_key(b'IA')
- assert 1 == d.has_key(b'OA')
+ self.assertFalse(d.has_key(b'bad key'))
+ self.assertTrue(d.has_key(b'IA'), d.keys())
+ self.assertTrue(d.has_key(b'OA'))
d.delete(b'IA')
del d[b'OA']
- assert 0 == d.has_key(b'IA')
- assert 0 == d.has_key(b'OA')
- assert len(d) == l-2
+ self.assertFalse(d.has_key(b'IA'))
+ self.assertFalse(d.has_key(b'OA'))
+ self.assertEqual(len(d), l-2)
values = []
for key in d.keys():
self.checkrec(key, value)
dbvalues = sorted(d.values(), key=lambda x: (str(type(x)), x))
- assert len(dbvalues) == len(d.keys())
+ self.assertEqual(len(dbvalues), len(d.keys()))
values.sort(key=lambda x: (str(type(x)), x))
- assert values == dbvalues, "%r != %r" % (values, dbvalues)
+ self.assertEqual(values, dbvalues, "%r != %r" % (values, dbvalues))
items = d.items()
- assert len(items) == len(values)
+ self.assertEqual(len(items), len(values))
for key, value in items:
self.checkrec(key, value)
- assert d.get(b'bad key') == None
- assert d.get(b'bad key', None) == None
- assert d.get(b'bad key', b'a string') == b'a string'
- assert d.get(b'bad key', [1, 2, 3]) == [1, 2, 3]
+ self.assertEqual(d.get(b'bad key'), None)
+ self.assertEqual(d.get(b'bad key', None), None)
+ self.assertEqual(d.get(b'bad key', b'a string'), b'a string')
+ self.assertEqual(d.get(b'bad key', [1, 2, 3]), [1, 2, 3])
d.set_get_returns_none(0)
self.assertRaises(db.DBNotFoundError, d.get, b'bad key')
d.set_get_returns_none(1)
d.put(b'new key', b'new data')
- assert d.get(b'new key') == b'new data'
- assert d[b'new key'] == b'new data'
+ self.assertEqual(d.get(b'new key'), b'new data')
+ self.assertEqual(d[b'new key'], b'new data')
rec = c.next()
del c
- assert count == len(d)
+ self.assertEqual(count, len(d))
count = 0
c = d.cursor()
self.checkrec(key, value)
rec = c.prev()
- assert count == len(d)
+ self.assertEqual(count, len(d))
c.set(b'SS')
key, value = c.current()
#----------------------------------------------------------------------
class BasicEnvShelveTestCase(DBShelveTestCase):
+ def setUp(self):
+ self.homeDir = tempfile.mkdtemp()
+ self.filename = 'dbshelve_db_file.db'
+ self.do_open()
+
def do_open(self):
- self.homeDir = homeDir = os.path.join(
- tempfile.gettempdir(), 'db_home')
- try: os.mkdir(homeDir)
- except os.error: pass
self.env = db.DBEnv()
- self.env.open(homeDir, self.envflags | db.DB_INIT_MPOOL | db.DB_CREATE)
+ self.env.open(self.homeDir, self.envflags | db.DB_INIT_MPOOL | db.DB_CREATE)
- self.filename = os.path.split(self.filename)[1]
self.d = dbshelve.DBShelf(self.env)
self.d.open(self.filename, self.dbtype, self.dbflags)
def tearDown(self):
self.do_close()
- import glob
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
# software has been tested, but no warranty is expressed or
# implied.
#
-# -- Gregory P. Smith <greg@electricrain.com>
+# -- Gregory P. Smith <greg@krypto.org>
#
# $Id$
+import shutil
import sys, os, re
import pickle
import tempfile
import unittest
-from .test_all import verbose
+from bsddb.test.test_all import verbose
try:
# For Pythons w/distutils pybsddb
#----------------------------------------------------------------------
class TableDBTestCase(unittest.TestCase):
- db_home = 'db_home'
db_name = 'test-table.db'
def setUp(self):
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- self.homeDir = homeDir
- try: os.mkdir(homeDir)
- except os.error: pass
+ self.homeDir = tempfile.mkdtemp()
self.tdb = dbtables.bsdTableDB(
- filename='tabletest.db', dbhome=homeDir, create=1)
+ filename='tabletest.db', dbhome=self.homeDir, create=1)
def tearDown(self):
self.tdb.close()
- import glob
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def test01(self):
tabname = "test01"
"""
import os
+import shutil
import sys
import tempfile
import glob
# For Python 2.3
from bsddb import db
-from .test_all import verbose
+from bsddb.test.test_all import verbose
# We're going to get warnings in this module about trying to close the db when
# its env is already closed. Let's just ignore those.
class DBEnvClosedEarlyCrash(unittest.TestCase):
def setUp(self):
- self.homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- try: os.mkdir(self.homeDir)
- except os.error: pass
+ self.homeDir = tempfile.mkdtemp()
+ old_tempfile_tempdir = tempfile.tempdir
tempfile.tempdir = self.homeDir
self.filename = os.path.split(tempfile.mktemp())[1]
- tempfile.tempdir = None
+ tempfile.tempdir = old_tempfile_tempdir
def tearDown(self):
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def test01_close_dbenv_before_db(self):
from bsddb import db
-from .test_all import verbose
+from bsddb.test.test_all import verbose
letters = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
"""TestCases for using the DB.join and DBCursor.join_item methods.
"""
+import shutil
import sys, os
import tempfile
import time
have_threads = 0
import unittest
-from .test_all import verbose
+from bsddb.test.test_all import verbose
from bsddb import db, dbshelve, StringKeys
def setUp(self):
self.filename = self.__class__.__name__ + '.db'
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- self.homeDir = homeDir
- try: os.mkdir(homeDir)
- except os.error: pass
+ self.homeDir = tempfile.mkdtemp()
self.env = db.DBEnv()
- self.env.open(homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOCK )
+ self.env.open(self.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)
+ shutil.rmtree(self.homeDir)
def test01_join(self):
if verbose:
TestCases for testing the locking sub-system.
"""
+import shutil
import sys, os
import tempfile
import time
import unittest
-from .test_all import verbose
+from bsddb.test.test_all import verbose
try:
# For Pythons w/distutils pybsddb
class LockingTestCase(unittest.TestCase):
def setUp(self):
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- self.homeDir = homeDir
- try: os.mkdir(homeDir)
- except os.error: pass
+ self.homeDir = tempfile.mkdtemp()
self.env = db.DBEnv()
- self.env.open(homeDir, db.DB_THREAD | db.DB_INIT_MPOOL |
+ self.env.open(self.homeDir, db.DB_THREAD | db.DB_INIT_MPOOL |
db.DB_INIT_LOCK | db.DB_CREATE)
def tearDown(self):
self.env.close()
- import glob
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def test01_simple(self):
"""
import os
+import shutil
import sys
import unittest
import tempfile
class MiscTestCase(unittest.TestCase):
def setUp(self):
self.filename = self.__class__.__name__ + '.db'
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- self.homeDir = homeDir
- try:
- os.mkdir(homeDir)
- except OSError:
- pass
+ self.homeDir = tempfile.mkdtemp()
def tearDown(self):
try:
os.remove(self.filename)
except OSError:
pass
- import glob
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def test01_badpointer(self):
dbs = dbshelve.open(self.filename)
+import shutil
import sys, os
import pickle
+import tempfile
import unittest
import glob
class pickleTestCase(unittest.TestCase):
"""Verify that DBError can be pickled and unpickled"""
- db_home = 'db_home'
db_name = 'test-dbobj.db'
def setUp(self):
- homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home')
- self.homeDir = homeDir
- try: os.mkdir(homeDir)
- except os.error: pass
+ self.homeDir = tempfile.mkdtemp()
def tearDown(self):
if hasattr(self, 'db'):
del self.db
if hasattr(self, 'env'):
del self.env
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def _base_test_pickle_DBError(self, pickle):
self.env = db.DBEnv()
self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL)
self.db = db.DB(self.env)
self.db.open(self.db_name, db.DB_HASH, db.DB_CREATE)
- self.db.put('spam', 'eggs')
- assert self.db['spam'] == 'eggs'
+ self.db.put(b'spam', b'eggs')
+ self.assertEqual(self.db[b'spam'], b'eggs')
try:
- self.db.put('spam', 'ham', flags=db.DB_NOOVERWRITE)
+ self.db.put(b'spam', b'ham', flags=db.DB_NOOVERWRITE)
except db.DBError as egg:
pickledEgg = pickle.dumps(egg)
#print repr(pickledEgg)
if rottenEgg.args != egg.args or type(rottenEgg) != type(egg):
raise Exception(rottenEgg, '!=', egg)
else:
- raise Exception("where's my DBError exception?!?")
+ self.fail("where's my DBError exception?!?")
self.db.close()
self.env.close()
# For Python 2.3
from bsddb import db
-from .test_all import verbose
+from bsddb.test.test_all import verbose
letters = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
"""
import os
+import shutil
import sys
import errno
import tempfile
from pprint import pprint
import unittest
-from .test_all import verbose
+from bsddb.test.test_all import verbose
try:
# For Pythons w/distutils pybsddb
class SimpleRecnoTestCase(unittest.TestCase):
def setUp(self):
self.filename = tempfile.mktemp()
+ self.homeDir = tempfile.mkdtemp()
def tearDown(self):
try:
os.remove(self.filename)
except OSError as e:
if e.errno != errno.EEXIST: raise
+ shutil.rmtree(self.homeDir)
def test01_basic(self):
d = db.DB()
just a line in the file, but you can set a different record delimiter
if needed.
"""
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- source = os.path.join(homeDir, 'test_recno.txt')
- if not os.path.isdir(homeDir):
- os.mkdir(homeDir)
+ source = os.path.join(self.homeDir, 'test_recno.txt')
f = open(source, 'w') # create the file
f.close()
import unittest
import os
+import shutil
import sys
import tempfile
import glob
except ImportError:
from bsddb import db
-from .test_all import verbose
+from bsddb.test.test_all import verbose
class DBSequenceTest(unittest.TestCase):
def setUp(self):
self.int_32_max = 0x100000000
- self.homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home')
- try:
- os.mkdir(self.homeDir)
- except os.error:
- pass
+ self.homeDir = tempfile.mkdtemp()
+ old_tempfile_tempdir = tempfile.tempdir
tempfile.tempdir = self.homeDir
self.filename = os.path.split(tempfile.mktemp())[1]
- tempfile.tempdir = None
+ tempfile.tempdir = old_tempfile_tempdir
self.dbenv = db.DBEnv()
self.dbenv.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL, 0o666)
self.dbenv.close()
del self.dbenv
- files = glob.glob(os.path.join(self.homeDir, '*'))
- for file in files:
- os.remove(file)
+ shutil.rmtree(self.homeDir)
def test_get(self):
self.seq = db.DBSequence(self.d, flags=0)
pass
import unittest
-from .test_all import verbose
+from bsddb.test.test_all import verbose
try:
# For Pythons w/distutils pybsddb
if verbose:
dbutils._deadlock_VerboseFile = sys.stdout
- homeDir = os.path.join(tempfile.gettempdir(), 'db_home')
- self.homeDir = homeDir
- try:
- os.mkdir(homeDir)
- except OSError as e:
- if e.errno != errno.EEXIST: raise
+ self.homeDir = tempfile.mkdtemp()
self.env = db.DBEnv()
self.setEnvOpts()
- self.env.open(homeDir, self.envflags | db.DB_CREATE)
+ self.env.open(self.homeDir, self.envflags | db.DB_CREATE)
self.filename = self.__class__.__name__ + '.db'
self.d = db.DB(self.env)
def makeData(self, key):
return DASH.join([key] * 5)
+ def _writerThread(self, *args, **kwargs):
+ raise RuntimeError("must override this in a subclass")
+
+ def _readerThread(self, *args, **kwargs):
+ raise RuntimeError("must override this in a subclass")
+
+ def writerThread(self, *args, **kwargs):
+ try:
+ self._writerThread(*args, **kwargs)
+ except db.DBLockDeadlockError:
+ if verbose:
+ print(currentThread().getName(), 'died from', e)
+
+ def readerThread(self, *args, **kwargs):
+ try:
+ self._readerThread(*args, **kwargs)
+ except db.DBLockDeadlockError as e:
+ if verbose:
+ print(currentThread().getName(), 'died from', e)
+
+
+
#----------------------------------------------------------------------
for t in threads:
t.join()
- def writerThread(self, d, howMany, writerNum):
+ def _writerThread(self, d, howMany, writerNum):
#time.sleep(0.01 * writerNum + 0.01)
name = currentThread().getName()
start = howMany * writerNum
for x in range(start, stop):
key = ('%04d' % x).encode("ascii")
dbutils.DeadlockWrap(d.put, key, self.makeData(key),
- max_retries=12)
+ max_retries=20)
if verbose and x % 100 == 0:
print("%s: records %d - %d finished" % (name, start, x))
if verbose:
print("%s: thread finished" % name)
- def readerThread(self, d, readerNum):
+ def _readerThread(self, d, readerNum):
time.sleep(0.01 * readerNum)
name = currentThread().getName()
for t in threads:
t.join()
- def writerThread(self, d, howMany, writerNum):
+ def _writerThread(self, d, howMany, writerNum):
name = currentThread().getName()
start = howMany * writerNum
stop = howMany * (writerNum + 1) - 1
for x in range(start, stop):
key = ('%04d' % x).encode("ascii")
dbutils.DeadlockWrap(d.put, key, self.makeData(key),
- max_retries=12)
+ max_retries=20)
if verbose and x % 100 == 0:
print("%s: records %d - %d finished" % (name, start, x))
if random() <= 0.05:
for y in range(start, x):
key = ('%04d' % x).encode("ascii")
- data = dbutils.DeadlockWrap(d.get, key, max_retries=12)
+ data = dbutils.DeadlockWrap(d.get, key, max_retries=20)
self.assertEqual(data, self.makeData(key))
# flush them
try:
- dbutils.DeadlockWrap(d.sync, max_retries=12)
+ dbutils.DeadlockWrap(d.sync, max_retries=20)
except db.DBIncompleteError as val:
if verbose:
print("could not complete sync()...")
# read them back, deleting a few
for x in range(start, stop):
key = ('%04d' % x).encode("ascii")
- data = dbutils.DeadlockWrap(d.get, key, max_retries=12)
+ data = dbutils.DeadlockWrap(d.get, key, max_retries=20)
if verbose and x % 100 == 0:
print("%s: fetched record (%s, %s)" % (name, key, data))
self.assertEqual(data, self.makeData(key))
if random() <= 0.10:
- dbutils.DeadlockWrap(d.delete, key, max_retries=12)
+ dbutils.DeadlockWrap(d.delete, key, max_retries=20)
if verbose:
print("%s: deleted record %s" % (name, key))
if verbose:
print("%s: thread finished" % name)
- def readerThread(self, d, readerNum):
+ def _readerThread(self, d, readerNum):
time.sleep(0.01 * readerNum)
name = currentThread().getName()
for loop in range(5):
c = d.cursor()
count = 0
- rec = dbutils.DeadlockWrap(c.first, max_retries=10)
+ rec = dbutils.DeadlockWrap(c.first, max_retries=20)
while rec:
count += 1
key, data = rec
self.assertEqual(self.makeData(key), data)
- rec = dbutils.DeadlockWrap(c.next, max_retries=10)
+ rec = dbutils.DeadlockWrap(c.next, max_retries=20)
if verbose:
print("%s: found %d records" % (name, count))
c.close()
txn.abort()
time.sleep(0.05)
- def writerThread(self, d, howMany, writerNum):
+ def _writerThread(self, d, howMany, writerNum):
name = currentThread().getName()
start = howMany * writerNum
stop = howMany * (writerNum + 1) - 1
if verbose:
print("%s: thread finished" % name)
- def readerThread(self, d, readerNum):
+ def _readerThread(self, d, readerNum):
time.sleep(0.01 * readerNum + 0.05)
name = currentThread().getName()
"""
import sys
import unittest
-from test.test_support import requires, verbose, run_unittest, unlink
+import test.test_support
+from test.test_support import requires, run_unittest, unlink
# When running as a script instead of within the regrtest framework, skip the
# requires test, since it's obvious we want to run them.
if __name__ != '__main__':
requires('bsddb')
-verbose = False
+import bsddb.test.test_all
if 'verbose' in sys.argv:
- verbose = True
+ bsddb.test.test_all.verbose = 1
sys.argv.remove('verbose')
if 'silent' in sys.argv: # take care of old flag, just in case
- verbose = False
+ bsddb.test.test_all.verbose = 0
sys.argv.remove('silent')
try:
# this is special, it used to segfault the interpreter
import bsddb.test.test_1413192
- except:
- for f in ['__db.001', '__db.002', '__db.003', 'log.0000000001']:
+ finally:
+ for f in ['xxx.db','__db.001','__db.002','__db.003','log.0000000001']:
unlink(f)
test_modules = [
'test_associate',
'test_basics',
'test_compat',
+ 'test_compare',
'test_dbobj',
'test_dbshelve',
'test_dbtables',
'test_join',
'test_lock',
'test_misc',
+ 'test_pickle',
'test_queue',
'test_recno',
'test_thread',
projects / python / commitdiff