Lib/bsddb/test/test_misc.py

   1 """Miscellaneous bsddb module test cases
   2 """
   3
   4 import os, sys
   5 import unittest
   6
   7 from test_all import db, dbshelve, hashopen, test_support, get_new_environment_path, get_new_database_path
   8
   9 #----------------------------------------------------------------------
  10
  11 class MiscTestCase(unittest.TestCase):
  12     if sys.version_info < (2, 4) :
  13         def assertTrue(self, expr, msg=None):
  14             self.failUnless(expr, msg=msg)
  15
  16         def assertFalse(self, expr, msg=None):
  17             self.failIf(expr, msg=msg)
  18
  19     def setUp(self):
  20         self.filename = get_new_database_path()
  21         self.homeDir = get_new_environment_path()
  22
  23     def tearDown(self):
  24         test_support.unlink(self.filename)
  25         test_support.rmtree(self.homeDir)
  26
  27     def test01_badpointer(self):
  28         dbs = dbshelve.open(self.filename)
  29         dbs.close()
  30         self.assertRaises(db.DBError, dbs.get, "foo")
  31
  32     def test02_db_home(self):
  33         env = db.DBEnv()
  34         # check for crash fixed when db_home is used before open()
  35         self.assert_(env.db_home is None)
  36         env.open(self.homeDir, db.DB_CREATE)
  37         if sys.version_info[0] < 3 :
  38             self.assertEqual(self.homeDir, env.db_home)
  39         else :
  40             self.assertEqual(bytes(self.homeDir, "ascii"), env.db_home)
  41
  42     def test03_repr_closed_db(self):
  43         db = hashopen(self.filename)
  44         db.close()
  45         rp = repr(db)
  46         self.assertEquals(rp, "{}")
  47
  48     def test04_repr_db(self) :
  49         db = hashopen(self.filename)
  50         d = {}
  51         for i in xrange(100) :
  52             db[repr(i)] = repr(100*i)
  53             d[repr(i)] = repr(100*i)
  54         db.close()
  55         db = hashopen(self.filename)
  56         rp = repr(db)
  57         self.assertEquals(rp, repr(d))
  58         db.close()
  59
  60     # http://sourceforge.net/tracker/index.php?func=detail&aid=1708868&group_id=13900&atid=313900
  61     #
  62     # See the bug report for details.
  63     #
  64     # The problem was that make_key_dbt() was not allocating a copy of
  65     # string keys but FREE_DBT() was always being told to free it when the
  66     # database was opened with DB_THREAD.
  67     def test05_double_free_make_key_dbt(self):
  68         try:
  69             db1 = db.DB()
  70             db1.open(self.filename, None, db.DB_BTREE,
  71                      db.DB_CREATE | db.DB_THREAD)
  72
  73             curs = db1.cursor()
  74             t = curs.get("/foo", db.DB_SET)
  75             # double free happened during exit from DBC_get
  76         finally:
  77             db1.close()
  78             test_support.unlink(self.filename)
  79
  80     def test06_key_with_null_bytes(self):
  81         try:
  82             db1 = db.DB()
  83             db1.open(self.filename, None, db.DB_HASH, db.DB_CREATE)
  84             db1['a'] = 'eh?'
  85             db1['a\x00'] = 'eh zed.'
  86             db1['a\x00a'] = 'eh zed eh?'
  87             db1['aaa'] = 'eh eh eh!'
  88             keys = db1.keys()
  89             keys.sort()
  90             self.assertEqual(['a', 'a\x00', 'a\x00a', 'aaa'], keys)
  91             self.assertEqual(db1['a'], 'eh?')
  92             self.assertEqual(db1['a\x00'], 'eh zed.')
  93             self.assertEqual(db1['a\x00a'], 'eh zed eh?')
  94             self.assertEqual(db1['aaa'], 'eh eh eh!')
  95         finally:
  96             db1.close()
  97             test_support.unlink(self.filename)
  98
  99     def test07_DB_set_flags_persists(self):
 100         if db.version() < (4,2):
 101             # The get_flags API required for this to work is only available
 102             # in Berkeley DB >= 4.2
 103             return
 104         try:
 105             db1 = db.DB()
 106             db1.set_flags(db.DB_DUPSORT)
 107             db1.open(self.filename, db.DB_HASH, db.DB_CREATE)
 108             db1['a'] = 'eh'
 109             db1['a'] = 'A'
 110             self.assertEqual([('a', 'A')], db1.items())
 111             db1.put('a', 'Aa')
 112             self.assertEqual([('a', 'A'), ('a', 'Aa')], db1.items())
 113             db1.close()
 114             db1 = db.DB()
 115             # no set_flags call, we're testing that it reads and obeys
 116             # the flags on open.
 117             db1.open(self.filename, db.DB_HASH)
 118             self.assertEqual([('a', 'A'), ('a', 'Aa')], db1.items())
 119             # if it read the flags right this will replace all values
 120             # for key 'a' instead of adding a new one.  (as a dict should)
 121             db1['a'] = 'new A'
 122             self.assertEqual([('a', 'new A')], db1.items())
 123         finally:
 124             db1.close()
 125             test_support.unlink(self.filename)
 126
 127
 128     def test08_ExceptionTypes(self) :
 129         self.assertTrue(issubclass(db.DBError, Exception))
 130         for i, j in db.__dict__.items() :
 131             if i.startswith("DB") and i.endswith("Error") :
 132                 self.assertTrue(issubclass(j, db.DBError), msg=i)
 133                 if i not in ("DBKeyEmptyError", "DBNotFoundError") :
 134                     self.assertFalse(issubclass(j, KeyError), msg=i)
 135
 136         # This two exceptions have two bases
 137         self.assertTrue(issubclass(db.DBKeyEmptyError, KeyError))
 138         self.assertTrue(issubclass(db.DBNotFoundError, KeyError))
 139
 140
 141 #----------------------------------------------------------------------
 142
 143
 144 def test_suite():
 145     return unittest.makeSuite(MiscTestCase)
 146
 147
 148 if __name__ == '__main__':
 149     unittest.main(defaultTest='test_suite')