| 1 | """TestCases for checking that it does not segfault when a DBEnv object |
| 2 | is closed before its DB objects. |
| 3 | """ |
| 4 | |
| 5 | import os |
| 6 | import sys |
| 7 | import tempfile |
| 8 | import glob |
| 9 | import unittest |
| 10 | |
| 11 | try: |
| 12 | # For Pythons w/distutils pybsddb |
| 13 | from bsddb3 import db |
| 14 | except ImportError: |
| 15 | # For Python 2.3 |
| 16 | from bsddb import db |
| 17 | |
| 18 | from test_all import verbose |
| 19 | |
| 20 | # We're going to get warnings in this module about trying to close the db when |
| 21 | # its env is already closed. Let's just ignore those. |
| 22 | try: |
| 23 | import warnings |
| 24 | except ImportError: |
| 25 | pass |
| 26 | else: |
| 27 | warnings.filterwarnings('ignore', |
| 28 | message='DB could not be closed in', |
| 29 | category=RuntimeWarning) |
| 30 | |
| 31 | |
| 32 | #---------------------------------------------------------------------- |
| 33 | |
| 34 | class DBEnvClosedEarlyCrash(unittest.TestCase): |
| 35 | def setUp(self): |
| 36 | self.homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home') |
| 37 | try: os.mkdir(self.homeDir) |
| 38 | except os.error: pass |
| 39 | tempfile.tempdir = self.homeDir |
| 40 | self.filename = os.path.split(tempfile.mktemp())[1] |
| 41 | tempfile.tempdir = None |
| 42 | |
| 43 | def tearDown(self): |
| 44 | files = glob.glob(os.path.join(self.homeDir, '*')) |
| 45 | for file in files: |
| 46 | os.remove(file) |
| 47 | |
| 48 | |
| 49 | def test01_close_dbenv_before_db(self): |
| 50 | dbenv = db.DBEnv() |
| 51 | dbenv.open(self.homeDir, |
| 52 | db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL, |
| 53 | 0666) |
| 54 | |
| 55 | d = db.DB(dbenv) |
| 56 | d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666) |
| 57 | |
| 58 | try: |
| 59 | dbenv.close() |
| 60 | except db.DBError: |
| 61 | try: |
| 62 | d.close() |
| 63 | except db.DBError: |
| 64 | return |
| 65 | assert 0, \ |
| 66 | "DB close did not raise an exception about its "\ |
| 67 | "DBEnv being trashed" |
| 68 | |
| 69 | # XXX This may fail when using older versions of BerkeleyDB. |
| 70 | # E.g. 3.2.9 never raised the exception. |
| 71 | assert 0, "dbenv did not raise an exception about its DB being open" |
| 72 | |
| 73 | |
| 74 | def test02_close_dbenv_delete_db_success(self): |
| 75 | dbenv = db.DBEnv() |
| 76 | dbenv.open(self.homeDir, |
| 77 | db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL, |
| 78 | 0666) |
| 79 | |
| 80 | d = db.DB(dbenv) |
| 81 | d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666) |
| 82 | |
| 83 | try: |
| 84 | dbenv.close() |
| 85 | except db.DBError: |
| 86 | pass # good, it should raise an exception |
| 87 | |
| 88 | del d |
| 89 | try: |
| 90 | import gc |
| 91 | except ImportError: |
| 92 | gc = None |
| 93 | if gc: |
| 94 | # force d.__del__ [DB_dealloc] to be called |
| 95 | gc.collect() |
| 96 | |
| 97 | |
| 98 | #---------------------------------------------------------------------- |
| 99 | |
| 100 | def test_suite(): |
| 101 | suite = unittest.TestSuite() |
| 102 | suite.addTest(unittest.makeSuite(DBEnvClosedEarlyCrash)) |
| 103 | return suite |
| 104 | |
| 105 | |
| 106 | if __name__ == '__main__': |
| 107 | unittest.main(defaultTest='test_suite') |