Lib/bsddb/test/test_early_close.py

   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, sys
   6 import unittest
   7
   8 from test_all import db, test_support, verbose, get_new_environment_path, get_new_database_path
   9
  10 # We're going to get warnings in this module about trying to close the db when
  11 # its env is already closed.  Let's just ignore those.
  12 try:
  13     import warnings
  14 except ImportError:
  15     pass
  16 else:
  17     warnings.filterwarnings('ignore',
  18                             message='DB could not be closed in',
  19                             category=RuntimeWarning)
  20
  21
  22 #----------------------------------------------------------------------
  23
  24 class DBEnvClosedEarlyCrash(unittest.TestCase):
  25     def setUp(self):
  26         self.homeDir = get_new_environment_path()
  27         self.filename = "test"
  28
  29     def tearDown(self):
  30         test_support.rmtree(self.homeDir)
  31
  32     def test01_close_dbenv_before_db(self):
  33         dbenv = db.DBEnv()
  34         dbenv.open(self.homeDir,
  35                    db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL,
  36                    0666)
  37
  38         d = db.DB(dbenv)
  39         d2 = db.DB(dbenv)
  40         d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
  41
  42         self.assertRaises(db.DBNoSuchFileError, d2.open,
  43                 self.filename+"2", db.DB_BTREE, db.DB_THREAD, 0666)
  44
  45         d.put("test","this is a test")
  46         self.assertEqual(d.get("test"), "this is a test", "put!=get")
  47         dbenv.close()  # This "close" should close the child db handle also
  48         self.assertRaises(db.DBError, d.get, "test")
  49
  50     def test02_close_dbenv_before_dbcursor(self):
  51         dbenv = db.DBEnv()
  52         dbenv.open(self.homeDir,
  53                    db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL,
  54                    0666)
  55
  56         d = db.DB(dbenv)
  57         d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
  58
  59         d.put("test","this is a test")
  60         d.put("test2","another test")
  61         d.put("test3","another one")
  62         self.assertEqual(d.get("test"), "this is a test", "put!=get")
  63         c=d.cursor()
  64         c.first()
  65         c.next()
  66         d.close()  # This "close" should close the child db handle also
  67      # db.close should close the child cursor
  68         self.assertRaises(db.DBError,c.next)
  69
  70         d = db.DB(dbenv)
  71         d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
  72         c=d.cursor()
  73         c.first()
  74         c.next()
  75         dbenv.close()
  76     # The "close" should close the child db handle also, with cursors
  77         self.assertRaises(db.DBError, c.next)
  78
  79     def test03_close_db_before_dbcursor_without_env(self):
  80         import os.path
  81         path=os.path.join(self.homeDir,self.filename)
  82         d = db.DB()
  83         d.open(path, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
  84
  85         d.put("test","this is a test")
  86         d.put("test2","another test")
  87         d.put("test3","another one")
  88         self.assertEqual(d.get("test"), "this is a test", "put!=get")
  89         c=d.cursor()
  90         c.first()
  91         c.next()
  92         d.close()
  93     # The "close" should close the child db handle also
  94         self.assertRaises(db.DBError, c.next)
  95
  96     def test04_close_massive(self):
  97         dbenv = db.DBEnv()
  98         dbenv.open(self.homeDir,
  99                    db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL,
 100                    0666)
 101
 102         dbs=[db.DB(dbenv) for i in xrange(16)]
 103         cursors=[]
 104         for i in dbs :
 105             i.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
 106
 107         dbs[10].put("test","this is a test")
 108         dbs[10].put("test2","another test")
 109         dbs[10].put("test3","another one")
 110         self.assertEqual(dbs[4].get("test"), "this is a test", "put!=get")
 111
 112         for i in dbs :
 113             cursors.extend([i.cursor() for j in xrange(32)])
 114
 115         for i in dbs[::3] :
 116             i.close()
 117         for i in cursors[::3] :
 118             i.close()
 119
 120     # Check for missing exception in DB! (after DB close)
 121         self.assertRaises(db.DBError, dbs[9].get, "test")
 122
 123     # Check for missing exception in DBCursor! (after DB close)
 124         self.assertRaises(db.DBError, cursors[101].first)
 125
 126         cursors[80].first()
 127         cursors[80].next()
 128         dbenv.close()  # This "close" should close the child db handle also
 129     # Check for missing exception! (after DBEnv close)
 130         self.assertRaises(db.DBError, cursors[80].next)
 131
 132     def test05_close_dbenv_delete_db_success(self):
 133         dbenv = db.DBEnv()
 134         dbenv.open(self.homeDir,
 135                    db.DB_INIT_CDB| db.DB_CREATE |db.DB_THREAD|db.DB_INIT_MPOOL,
 136                    0666)
 137
 138         d = db.DB(dbenv)
 139         d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
 140
 141         dbenv.close()  # This "close" should close the child db handle also
 142
 143         del d
 144         try:
 145             import gc
 146         except ImportError:
 147             gc = None
 148         if gc:
 149             # force d.__del__ [DB_dealloc] to be called
 150             gc.collect()
 151
 152     def test06_close_txn_before_dup_cursor(self) :
 153         dbenv = db.DBEnv()
 154         dbenv.open(self.homeDir,db.DB_INIT_TXN | db.DB_INIT_MPOOL |
 155                 db.DB_INIT_LOG | db.DB_CREATE)
 156         d = db.DB(dbenv)
 157         txn = dbenv.txn_begin()
 158         d.open(self.filename, dbtype = db.DB_HASH, flags = db.DB_CREATE,
 159                 txn=txn)
 160         d.put("XXX", "yyy", txn=txn)
 161         txn.commit()
 162         txn = dbenv.txn_begin()
 163         c1 = d.cursor(txn)
 164         c2 = c1.dup()
 165         self.assertEquals(("XXX", "yyy"), c1.first())
 166
 167         # Not interested in warnings about implicit close.
 168         import warnings
 169         if sys.version_info < (2, 6) :
 170             # Completely resetting the warning state is
 171             # problematic with python >=2.6 with -3 (py3k warning),
 172             # because some stdlib modules selectively ignores warnings.
 173             warnings.simplefilter("ignore")
 174             txn.commit()
 175             warnings.resetwarnings()
 176         else :
 177             # When we drop support for python 2.3 and 2.4
 178             # we could use: (in 2.5 we need a __future__ statement)
 179             #
 180             #    with warnings.catch_warnings():
 181             #        warnings.simplefilter("ignore")
 182             #        txn.commit()
 183             #
 184             # We can not use "with" as is, because it would be invalid syntax
 185             # in python 2.3, 2.4 and (with no __future__) 2.5.
 186             # Here we simulate "with" following PEP 343 :
 187             w = warnings.catch_warnings()
 188             w.__enter__()
 189             try :
 190                 warnings.simplefilter("ignore")
 191                 txn.commit()
 192             finally :
 193                 w.__exit__()
 194
 195         self.assertRaises(db.DBCursorClosedError, c2.first)
 196
 197     if db.version() > (4,3,0) :
 198         def test07_close_db_before_sequence(self):
 199             import os.path
 200             path=os.path.join(self.homeDir,self.filename)
 201             d = db.DB()
 202             d.open(path, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 0666)
 203             dbs=db.DBSequence(d)
 204             d.close()  # This "close" should close the child DBSequence also
 205             dbs.close()  # If not closed, core dump (in Berkeley DB 4.6.*)
 206
 207 #----------------------------------------------------------------------
 208
 209 def test_suite():
 210     suite = unittest.TestSuite()
 211     suite.addTest(unittest.makeSuite(DBEnvClosedEarlyCrash))
 212     return suite
 213
 214
 215 if __name__ == '__main__':
 216     unittest.main(defaultTest='test_suite')