| """TestCases for checking that it does not segfault when a DBEnv object |
| is closed before its DB objects. |
| """ |
| |
| import os |
| import shutil |
| import sys |
| import tempfile |
| import glob |
| import unittest |
| |
| try: |
| # For Pythons w/distutils pybsddb |
| from bsddb3 import db |
| except ImportError: |
| # For Python 2.3 |
| from bsddb import db |
| |
| 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. |
| 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 = tempfile.mkdtemp() |
| old_tempfile_tempdir = tempfile.tempdir |
| tempfile.tempdir = self.homeDir |
| self.filename = os.path.split(tempfile.mktemp())[1] |
| tempfile.tempdir = old_tempfile_tempdir |
| |
| def tearDown(self): |
| shutil.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, |
| 0o666) |
| |
| d = db.DB(dbenv) |
| d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0o666) |
| |
| 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, |
| 0o666) |
| |
| d = db.DB(dbenv) |
| d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0o666) |
| |
| 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') |