| """TestCases for checking that it does not segfault when a DBEnv object |
| is closed before its DB objects. |
| """ |
| |
| import os, sys |
| import unittest |
| |
| from test_all import db, test_support, verbose, get_new_environment_path, get_new_database_path |
| |
| # 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 = get_new_environment_path() |
| self.filename = "test" |
| |
| 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) |
| d2 = db.DB(dbenv) |
| d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666) |
| |
| self.assertRaises(db.DBNoSuchFileError, d2.open, |
| self.filename+"2", db.DB_BTREE, db.DB_THREAD, 0666) |
| |
| d.put("test","this is a test") |
| self.assertEqual(d.get("test"), "this is a test", "put!=get") |
| dbenv.close() # This "close" should close the child db handle also |
| self.assertRaises(db.DBError, d.get, "test") |
| |
| def test02_close_dbenv_before_dbcursor(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) |
| |
| d.put("test","this is a test") |
| d.put("test2","another test") |
| d.put("test3","another one") |
| self.assertEqual(d.get("test"), "this is a test", "put!=get") |
| c=d.cursor() |
| c.first() |
| c.next() |
| d.close() # This "close" should close the child db handle also |
| # db.close should close the child cursor |
| self.assertRaises(db.DBError,c.next) |
| |
| d = db.DB(dbenv) |
| d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666) |
| c=d.cursor() |
| c.first() |
| c.next() |
| dbenv.close() |
| # The "close" should close the child db handle also, with cursors |
| self.assertRaises(db.DBError, c.next) |
| |
| def test03_close_db_before_dbcursor_without_env(self): |
| import os.path |
| path=os.path.join(self.homeDir,self.filename) |
| d = db.DB() |
| d.open(path, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666) |
| |
| d.put("test","this is a test") |
| d.put("test2","another test") |
| d.put("test3","another one") |
| self.assertEqual(d.get("test"), "this is a test", "put!=get") |
| c=d.cursor() |
| c.first() |
| c.next() |
| d.close() |
| # The "close" should close the child db handle also |
| self.assertRaises(db.DBError, c.next) |
| |
| def test04_close_massive(self): |
| dbenv = db.DBEnv() |
| dbenv.open(self.homeDir, |
| db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL, |
| 0666) |
| |
| dbs=[db.DB(dbenv) for i in xrange(16)] |
| cursors=[] |
| for i in dbs : |
| i.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666) |
| |
| dbs[10].put("test","this is a test") |
| dbs[10].put("test2","another test") |
| dbs[10].put("test3","another one") |
| self.assertEqual(dbs[4].get("test"), "this is a test", "put!=get") |
| |
| for i in dbs : |
| cursors.extend([i.cursor() for j in xrange(32)]) |
| |
| for i in dbs[::3] : |
| i.close() |
| for i in cursors[::3] : |
| i.close() |
| |
| # Check for missing exception in DB! (after DB close) |
| self.assertRaises(db.DBError, dbs[9].get, "test") |
| |
| # Check for missing exception in DBCursor! (after DB close) |
| self.assertRaises(db.DBError, cursors[101].first) |
| |
| cursors[80].first() |
| cursors[80].next() |
| dbenv.close() # This "close" should close the child db handle also |
| # Check for missing exception! (after DBEnv close) |
| self.assertRaises(db.DBError, cursors[80].next) |
| |
| def test05_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) |
| |
| dbenv.close() # This "close" should close the child db handle also |
| |
| del d |
| try: |
| import gc |
| except ImportError: |
| gc = None |
| if gc: |
| # force d.__del__ [DB_dealloc] to be called |
| gc.collect() |
| |
| def test06_close_txn_before_dup_cursor(self) : |
| dbenv = db.DBEnv() |
| dbenv.open(self.homeDir,db.DB_INIT_TXN | db.DB_INIT_MPOOL | |
| db.DB_INIT_LOG | db.DB_CREATE) |
| d = db.DB(dbenv) |
| txn = dbenv.txn_begin() |
| d.open(self.filename, dbtype = db.DB_HASH, flags = db.DB_CREATE, |
| txn=txn) |
| d.put("XXX", "yyy", txn=txn) |
| txn.commit() |
| txn = dbenv.txn_begin() |
| c1 = d.cursor(txn) |
| c2 = c1.dup() |
| self.assertEqual(("XXX", "yyy"), c1.first()) |
| |
| # Not interested in warnings about implicit close. |
| import warnings |
| if sys.version_info < (2, 6) : |
| # Completely resetting the warning state is |
| # problematic with python >=2.6 with -3 (py3k warning), |
| # because some stdlib modules selectively ignores warnings. |
| warnings.simplefilter("ignore") |
| txn.commit() |
| warnings.resetwarnings() |
| else : |
| # When we drop support for python 2.4 |
| # we could use: (in 2.5 we need a __future__ statement) |
| # |
| # with warnings.catch_warnings(): |
| # warnings.simplefilter("ignore") |
| # txn.commit() |
| # |
| # We can not use "with" as is, because it would be invalid syntax |
| # in python 2.4 and (with no __future__) 2.5. |
| # Here we simulate "with" following PEP 343 : |
| w = warnings.catch_warnings() |
| w.__enter__() |
| try : |
| warnings.simplefilter("ignore") |
| txn.commit() |
| finally : |
| w.__exit__() |
| |
| self.assertRaises(db.DBCursorClosedError, c2.first) |
| |
| def test07_close_db_before_sequence(self): |
| import os.path |
| path=os.path.join(self.homeDir,self.filename) |
| d = db.DB() |
| d.open(path, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666) |
| dbs=db.DBSequence(d) |
| d.close() # This "close" should close the child DBSequence also |
| dbs.close() # If not closed, core dump (in Berkeley DB 4.6.*) |
| |
| #---------------------------------------------------------------------- |
| |
| def test_suite(): |
| suite = unittest.TestSuite() |
| suite.addTest(unittest.makeSuite(DBEnvClosedEarlyCrash)) |
| return suite |
| |
| |
| if __name__ == '__main__': |
| unittest.main(defaultTest='test_suite') |
