mirror of https://github.com/python/cpython
108 lines
2.9 KiB
Python
108 lines
2.9 KiB
Python
"""TestCases for checking that it does not segfault when a DBEnv object
|
|
is closed before its DB objects.
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
try:
|
|
# For Pythons w/distutils pybsddb
|
|
from bsddb3 import db
|
|
except ImportError:
|
|
# For Python 2.3
|
|
from bsddb import db
|
|
|
|
try:
|
|
from bsddb3 import test_support
|
|
except ImportError:
|
|
from test import test_support
|
|
|
|
from 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.
|
|
try:
|
|
import warnings
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
warnings.filterwarnings('ignore',
|
|
message='DB could not be closed in',
|
|
category=RuntimeWarning)
|
|
|
|
|
|
#----------------------------------------------------------------------
|
|
|
|
class DBEnvClosedEarlyCrash(unittest.TestCase):
|
|
def setUp(self):
|
|
self.homeDir = os.path.join(tempfile.gettempdir(), 'db_home%d'%os.getpid())
|
|
try: os.mkdir(self.homeDir)
|
|
except os.error: pass
|
|
tempfile.tempdir = self.homeDir
|
|
self.filename = os.path.split(tempfile.mktemp())[1]
|
|
tempfile.tempdir = None
|
|
|
|
def tearDown(self):
|
|
test_support.rmtree(self.homeDir)
|
|
|
|
def test01_close_dbenv_before_db(self):
|
|
dbenv = db.DBEnv()
|
|
dbenv.open(self.homeDir,
|
|
db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL,
|
|
0666)
|
|
|
|
d = db.DB(dbenv)
|
|
d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
|
|
|
|
try:
|
|
dbenv.close()
|
|
except db.DBError:
|
|
try:
|
|
d.close()
|
|
except db.DBError:
|
|
return
|
|
assert 0, \
|
|
"DB close did not raise an exception about its "\
|
|
"DBEnv being trashed"
|
|
|
|
# XXX This may fail when using older versions of BerkeleyDB.
|
|
# E.g. 3.2.9 never raised the exception.
|
|
assert 0, "dbenv did not raise an exception about its DB being open"
|
|
|
|
|
|
def test02_close_dbenv_delete_db_success(self):
|
|
dbenv = db.DBEnv()
|
|
dbenv.open(self.homeDir,
|
|
db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL,
|
|
0666)
|
|
|
|
d = db.DB(dbenv)
|
|
d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
|
|
|
|
try:
|
|
dbenv.close()
|
|
except db.DBError:
|
|
pass # good, it should raise an exception
|
|
|
|
del d
|
|
try:
|
|
import gc
|
|
except ImportError:
|
|
gc = None
|
|
if gc:
|
|
# force d.__del__ [DB_dealloc] to be called
|
|
gc.collect()
|
|
|
|
|
|
#----------------------------------------------------------------------
|
|
|
|
def test_suite():
|
|
suite = unittest.TestSuite()
|
|
suite.addTest(unittest.makeSuite(DBEnvClosedEarlyCrash))
|
|
return suite
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(defaultTest='test_suite')
|