text: """
  Copyright (c) 2008-2014, Jesus Cea Avion <jcea@jcea.es>
  All rights reserved.
  
  Redistribution and use in source and binary forms, with or without
  modification, are permitted provided that the following conditions
  are met:
  
      1. Redistributions of source code must retain the above copyright
      notice, this list of conditions and the following disclaimer.
  
      2. Redistributions in binary form must reproduce the above
      copyright notice, this list of conditions and the following
      disclaimer in the documentation and/or other materials provided
      with the distribution.
  
      3. Neither the name of Jesus Cea Avion nor the names of its
      contributors may be used to endorse or promote products derived
      from this software without specific prior written permission.
  
      THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
      CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
      INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
      MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
      DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS
      BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
      EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
              TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
              DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
      ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR
      TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
      THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
      SUCH DAMAGE.
      """
  
  import unittest
  import os, glob
  
  from test_all import db, test_support, get_new_environment_path, \
          get_new_database_path
  
  #----------------------------------------------------------------------
  
  class DBEnv(unittest.TestCase):
      def setUp(self):
          self.homeDir = get_new_environment_path()
          self.env = db.DBEnv()
  
      def tearDown(self):
          self.env.close()
          del self.env
          test_support.rmtree(self.homeDir)
  
  class DBEnv_general(DBEnv) :
      def test_get_open_flags(self) :
          flags = db.DB_CREATE | db.DB_INIT_MPOOL
          self.env.open(self.homeDir, flags)
          self.assertEqual(flags, self.env.get_open_flags())
  
      def test_get_open_flags2(self) :
          flags = db.DB_CREATE | db.DB_INIT_MPOOL | \
                  db.DB_INIT_LOCK | db.DB_THREAD
          self.env.open(self.homeDir, flags)
          self.assertEqual(flags, self.env.get_open_flags())
  
      def test_lk_partitions(self) :
          for i in [10, 20, 40] :
              self.env.set_lk_partitions(i)
              self.assertEqual(i, self.env.get_lk_partitions())
  
      def test_getset_intermediate_dir_mode(self) :
          self.assertEqual(None, self.env.get_intermediate_dir_mode())
          for mode in ["rwx------", "rw-rw-rw-", "rw-r--r--"] :
              self.env.set_intermediate_dir_mode(mode)
              self.assertEqual(mode, self.env.get_intermediate_dir_mode())
          self.assertRaises(db.DBInvalidArgError,
                  self.env.set_intermediate_dir_mode, "abcde")
  
      def test_thread(self) :
          for i in [16, 100, 1000] :
              self.env.set_thread_count(i)
              self.assertEqual(i, self.env.get_thread_count())
  
      def test_cache_max(self) :
          for size in [64, 128] :
              size = size*1024*1024  # Megabytes
              self.env.set_cache_max(0, size)
              size2 = self.env.get_cache_max()
              self.assertEqual(0, size2[0])
              self.assertTrue(size <= size2[1])
              self.assertTrue(2*size > size2[1])
  
      def test_mutex_stat(self) :
          self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL |
                  db.DB_INIT_LOCK)
          stat = self.env.mutex_stat()
          self.assertTrue("mutex_inuse_max" in stat)
  
      def test_lg_filemode(self) :
          for i in [0600, 0660, 0666] :
              self.env.set_lg_filemode(i)
              self.assertEqual(i, self.env.get_lg_filemode())
  
      def test_mp_max_openfd(self) :
          for i in [17, 31, 42] :
              self.env.set_mp_max_openfd(i)
              self.assertEqual(i, self.env.get_mp_max_openfd())
  
      def test_mp_max_write(self) :
          for i in [100, 200, 300] :
              for j in [1, 2, 3] :
                  j *= 1000000
                  self.env.set_mp_max_write(i, j)
                  v=self.env.get_mp_max_write()
                  self.assertEqual((i, j), v)
  
      def test_invalid_txn(self) :
          # This environment doesn't support transactions
          self.assertRaises(db.DBInvalidArgError, self.env.txn_begin)
  
      def test_mp_mmapsize(self) :
          for i in [16, 32, 64] :
              i *= 1024*1024
              self.env.set_mp_mmapsize(i)
              self.assertEqual(i, self.env.get_mp_mmapsize())
  
      def test_tmp_dir(self) :
          for i in ["a", "bb", "ccc"] :
              self.env.set_tmp_dir(i)
              self.assertEqual(i, self.env.get_tmp_dir())
  
      def test_flags(self) :
          self.env.set_flags(db.DB_AUTO_COMMIT, 1)
          self.assertEqual(db.DB_AUTO_COMMIT, self.env.get_flags())
          self.env.set_flags(db.DB_TXN_NOSYNC, 1)
          self.assertEqual(db.DB_AUTO_COMMIT | db.DB_TXN_NOSYNC,
                  self.env.get_flags())
          self.env.set_flags(db.DB_AUTO_COMMIT, 0)
          self.assertEqual(db.DB_TXN_NOSYNC, self.env.get_flags())
          self.env.set_flags(db.DB_TXN_NOSYNC, 0)
          self.assertEqual(0, self.env.get_flags())
  
      def test_lk_max_objects(self) :
          for i in [1000, 2000, 3000] :
              self.env.set_lk_max_objects(i)
              self.assertEqual(i, self.env.get_lk_max_objects())
  
      def test_lk_max_locks(self) :
          for i in [1000, 2000, 3000] :
              self.env.set_lk_max_locks(i)
              self.assertEqual(i, self.env.get_lk_max_locks())
  
      def test_lk_max_lockers(self) :
          for i in [1000, 2000, 3000] :
              self.env.set_lk_max_lockers(i)
              self.assertEqual(i, self.env.get_lk_max_lockers())
  
      def test_lg_regionmax(self) :
          for i in [128, 256, 1000] :
              i = i*1024*1024
              self.env.set_lg_regionmax(i)
              j = self.env.get_lg_regionmax()
              self.assertTrue(i <= j)
              self.assertTrue(2*i > j)
  
      def test_lk_detect(self) :
          flags= [db.DB_LOCK_DEFAULT, db.DB_LOCK_EXPIRE, db.DB_LOCK_MAXLOCKS,
                  db.DB_LOCK_MINLOCKS, db.DB_LOCK_MINWRITE,
                  db.DB_LOCK_OLDEST, db.DB_LOCK_RANDOM, db.DB_LOCK_YOUNGEST]
  
          flags.append(db.DB_LOCK_MAXWRITE)
  
          for i in flags :
              self.env.set_lk_detect(i)
              self.assertEqual(i, self.env.get_lk_detect())
  
      def test_lg_dir(self) :
          for i in ["a", "bb", "ccc", "dddd"] :
              self.env.set_lg_dir(i)
              self.assertEqual(i, self.env.get_lg_dir())
  
      def test_lg_bsize(self) :
          log_size = 70*1024
          self.env.set_lg_bsize(log_size)
          self.assertTrue(self.env.get_lg_bsize() >= log_size)
          self.assertTrue(self.env.get_lg_bsize() < 4*log_size)
          self.env.set_lg_bsize(4*log_size)
          self.assertTrue(self.env.get_lg_bsize() >= 4*log_size)
  
      def test_setget_data_dirs(self) :
          dirs = ("a", "b", "c", "d")
          for i in dirs :
              self.env.set_data_dir(i)
          self.assertEqual(dirs, self.env.get_data_dirs())
  
      def test_setget_cachesize(self) :
          cachesize = (0, 512*1024*1024, 3)
          self.env.set_cachesize(*cachesize)
          self.assertEqual(cachesize, self.env.get_cachesize())
  
          cachesize = (0, 1*1024*1024, 5)
          self.env.set_cachesize(*cachesize)
          cachesize2 = self.env.get_cachesize()
          self.assertEqual(cachesize[0], cachesize2[0])
          self.assertEqual(cachesize[2], cachesize2[2])
          # Berkeley DB expands the cache 25% accounting overhead,
          # if the cache is small.
          self.assertEqual(125, int(100.0*cachesize2[1]/cachesize[1]))
  
          # You can not change configuration after opening
          # the environment.
          self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL)
          cachesize = (0, 2*1024*1024, 1)
          self.assertRaises(db.DBInvalidArgError,
              self.env.set_cachesize, *cachesize)
          cachesize3 = self.env.get_cachesize()
          self.assertEqual(cachesize2[0], cachesize3[0])
          self.assertEqual(cachesize2[2], cachesize3[2])
          # In Berkeley DB 5.1, the cachesize can change when opening the Env
          self.assertTrue(cachesize2[1] <= cachesize3[1])
  
      def test_set_cachesize_dbenv_db(self) :
          # You can not configure the cachesize using
          # the database handle, if you are using an environment.
          d = db.DB(self.env)
          self.assertRaises(db.DBInvalidArgError,
              d.set_cachesize, 0, 1024*1024, 1)
  
      def test_setget_shm_key(self) :
          shm_key=137
          self.env.set_shm_key(shm_key)
          self.assertEqual(shm_key, self.env.get_shm_key())
          self.env.set_shm_key(shm_key+1)
          self.assertEqual(shm_key+1, self.env.get_shm_key())
  
          # You can not change configuration after opening
          # the environment.
          self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL)
          # If we try to reconfigure cache after opening the
          # environment, core dump.
          self.assertRaises(db.DBInvalidArgError,
              self.env.set_shm_key, shm_key)
          self.assertEqual(shm_key+1, self.env.get_shm_key())
  
      def test_mutex_setget_max(self) :
          v = self.env.mutex_get_max()
          v2 = v*2+1
  
          self.env.mutex_set_max(v2)
          self.assertEqual(v2, self.env.mutex_get_max())
  
          self.env.mutex_set_max(v)
          self.assertEqual(v, self.env.mutex_get_max())
  
          # You can not change configuration after opening
          # the environment.
          self.env.open(self.homeDir, db.DB_CREATE)
          self.assertRaises(db.DBInvalidArgError,
                  self.env.mutex_set_max, v2)
  
      def test_mutex_setget_increment(self) :
          v = self.env.mutex_get_increment()
          v2 = 127
  
          self.env.mutex_set_increment(v2)
          self.assertEqual(v2, self.env.mutex_get_increment())
  
          self.env.mutex_set_increment(v)
          self.assertEqual(v, self.env.mutex_get_increment())
  
          # You can not change configuration after opening
          # the environment.
          self.env.open(self.homeDir, db.DB_CREATE)
          self.assertRaises(db.DBInvalidArgError,
                  self.env.mutex_set_increment, v2)
  
      def test_mutex_setget_tas_spins(self) :
          self.env.mutex_set_tas_spins(0)  # Default = BDB decides
          v = self.env.mutex_get_tas_spins()
          v2 = v*2+1
  
          self.env.mutex_set_tas_spins(v2)
          self.assertEqual(v2, self.env.mutex_get_tas_spins())
  
          self.env.mutex_set_tas_spins(v)
          self.assertEqual(v, self.env.mutex_get_tas_spins())
  
          # In this case, you can change configuration
          # after opening the environment.
          self.env.open(self.homeDir, db.DB_CREATE)
          self.env.mutex_set_tas_spins(v2)
  
      def test_mutex_setget_align(self) :
          v = self.env.mutex_get_align()
          v2 = 64
          if v == 64 :
              v2 = 128
  
          self.env.mutex_set_align(v2)
          self.assertEqual(v2, self.env.mutex_get_align())
  
          # Requires a nonzero power of two
          self.assertRaises(db.DBInvalidArgError,
                  self.env.mutex_set_align, 0)
          self.assertRaises(db.DBInvalidArgError,
                  self.env.mutex_set_align, 17)
  
          self.env.mutex_set_align(2*v2)
          self.assertEqual(2*v2, self.env.mutex_get_align())
  
          # You can not change configuration after opening
          # the environment.
          self.env.open(self.homeDir, db.DB_CREATE)
          self.assertRaises(db.DBInvalidArgError,
                  self.env.mutex_set_align, v2)
  
  
  class DBEnv_log(DBEnv) :
      def setUp(self):
          DBEnv.setUp(self)
          self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOG)
  
      def test_log_file(self) :
          log_file = self.env.log_file((1, 1))
          self.assertEqual("log.0000000001", log_file[-14:])
  
      # The version with transactions is checked in other test object
      def test_log_printf(self) :
          msg = "This is a test..."
          self.env.log_printf(msg)
          logc = self.env.log_cursor()
          self.assertTrue(msg in (logc.last()[1]))
  
      if db.version() >= (4, 7) :
          def test_log_config(self) :
              self.env.log_set_config(db.DB_LOG_DSYNC | db.DB_LOG_ZERO, 1)
              self.assertTrue(self.env.log_get_config(db.DB_LOG_DSYNC))
              self.assertTrue(self.env.log_get_config(db.DB_LOG_ZERO))
              self.env.log_set_config(db.DB_LOG_ZERO, 0)
              self.assertTrue(self.env.log_get_config(db.DB_LOG_DSYNC))
              self.assertFalse(self.env.log_get_config(db.DB_LOG_ZERO))
  
  
  class DBEnv_log_txn(DBEnv) :
      def setUp(self):
          DBEnv.setUp(self)
          self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL |
                  db.DB_INIT_LOG | db.DB_INIT_TXN)
  
      if db.version() < (5, 2) :
          def test_tx_max(self) :
              txns=[]
              def tx() :
                  for i in xrange(self.env.get_tx_max()) :
                      txns.append(self.env.txn_begin())
  
              tx()
              self.assertRaises(MemoryError, tx)
  
              # Abort the transactions before garbage collection,
              # to avoid "warnings".
              for i in txns :
                  i.abort()
  
      # The version without transactions is checked in other test object
      def test_log_printf(self) :
          msg = "This is a test..."
          txn = self.env.txn_begin()
          self.env.log_printf(msg, txn=txn)
          txn.commit()
          logc = self.env.log_cursor()
          logc.last()  # Skip the commit
          self.assertTrue(msg in (logc.prev()[1]))
  
          msg = "This is another test..."
          txn = self.env.txn_begin()
          self.env.log_printf(msg, txn=txn)
          txn.abort()  # Do not store the new message
          logc.last()  # Skip the abort
          self.assertTrue(msg not in (logc.prev()[1]))
  
          msg = "This is a third test..."
          txn = self.env.txn_begin()
          self.env.log_printf(msg, txn=txn)
          txn.commit()  # Do not store the new message
          logc.last()  # Skip the commit
          self.assertTrue(msg in (logc.prev()[1]))
  
  
  class DBEnv_memp(DBEnv):
      def setUp(self):
          DBEnv.setUp(self)
          self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL | db.DB_INIT_LOG)
          self.db = db.DB(self.env)
          self.db.open("test", db.DB_HASH, db.DB_CREATE, 0660)
  
      def tearDown(self):
          self.db.close()
          del self.db
          DBEnv.tearDown(self)
  
      def test_memp_1_trickle(self) :
          self.db.put("hi", "bye")
          self.assertTrue(self.env.memp_trickle(100) > 0)
  
  # Preserve the order, do "memp_trickle" test first
      def test_memp_2_sync(self) :
          self.db.put("hi", "bye")
          self.env.memp_sync()  # Full flush
          # Nothing to do...
          self.assertTrue(self.env.memp_trickle(100) == 0)
  
          self.db.put("hi", "bye2")
          self.env.memp_sync((1, 0))  # NOP, probably
          # Something to do... or not
          self.assertTrue(self.env.memp_trickle(100) >= 0)
  
          self.db.put("hi", "bye3")
          self.env.memp_sync((123, 99))  # Full flush
          # Nothing to do...
          self.assertTrue(self.env.memp_trickle(100) == 0)
  
      def test_memp_stat_1(self) :
          stats = self.env.memp_stat()  # No param
          self.assertTrue(len(stats)==2)
          self.assertTrue("cache_miss" in stats[0])
          stats = self.env.memp_stat(db.DB_STAT_CLEAR)  # Positional param
          self.assertTrue("cache_miss" in stats[0])
          stats = self.env.memp_stat(flags=0)  # Keyword param
          self.assertTrue("cache_miss" in stats[0])
  
      def test_memp_stat_2(self) :
          stats=self.env.memp_stat()[1]
          self.assertTrue(len(stats))==1
          self.assertTrue("test" in stats)
          self.assertTrue("page_in" in stats["test"])
  
  class DBEnv_logcursor(DBEnv):
      def setUp(self):
          DBEnv.setUp(self)
          self.env.open(self.homeDir, db.DB_CREATE | db.DB_INIT_MPOOL |
                  db.DB_INIT_LOG | db.DB_INIT_TXN)
          txn = self.env.txn_begin()
          self.db = db.DB(self.env)
          self.db.open("test", db.DB_HASH, db.DB_CREATE, 0660, txn=txn)
          txn.commit()
          for i in ["2", "8", "20"] :
              txn = self.env.txn_begin()
              self.db.put(key = i, data = i*int(i), txn=txn)
              txn.commit()
  
      def tearDown(self):
          self.db.close()
          del self.db
          DBEnv.tearDown(self)
  
      def _check_return(self, value) :
          self.assertTrue(isinstance(value, tuple))
          self.assertEqual(len(value), 2)
          self.assertTrue(isinstance(value[0], tuple))
          self.assertEqual(len(value[0]), 2)
          self.assertTrue(isinstance(value[0][0], int))
          self.assertTrue(isinstance(value[0][1], int))
          self.assertTrue(isinstance(value[1], str))
  
      # Preserve test order
      def test_1_first(self) :
          logc = self.env.log_cursor()
          v = logc.first()
          self._check_return(v)
          self.assertTrue((1, 1) < v[0])
          self.assertTrue(len(v[1])>0)
  
      def test_2_last(self) :
          logc = self.env.log_cursor()
          lsn_first = logc.first()[0]
          v = logc.last()
          self._check_return(v)
          self.assertTrue(lsn_first < v[0])
  
      def test_3_next(self) :
          logc = self.env.log_cursor()
          lsn_last = logc.last()[0]
          self.assertEqual(logc.next(), None)
          lsn_first = logc.first()[0]
          v = logc.next()
          self._check_return(v)
          self.assertTrue(lsn_first < v[0])
          self.assertTrue(lsn_last > v[0])
  
          v2 = logc.next()
          self.assertTrue(v2[0] > v[0])
          self.assertTrue(lsn_last > v2[0])
  
          v3 = logc.next()
          self.assertTrue(v3[0] > v2[0])
          self.assertTrue(lsn_last > v3[0])
  
      def test_4_prev(self) :
          logc = self.env.log_cursor()
          lsn_first = logc.first()[0]
          self.assertEqual(logc.prev(), None)
          lsn_last = logc.last()[0]
          v = logc.prev()
          self._check_return(v)
          self.assertTrue(lsn_first < v[0])
          self.assertTrue(lsn_last > v[0])
  
          v2 = logc.prev()
          self.assertTrue(v2[0] < v[0])
          self.assertTrue(lsn_first < v2[0])
  
          v3 = logc.prev()
          self.assertTrue(v3[0] < v2[0])
          self.assertTrue(lsn_first < v3[0])
  
      def test_5_current(self) :
          logc = self.env.log_cursor()
          logc.first()
          v = logc.next()
          self.assertEqual(v, logc.current())
  
      def test_6_set(self) :
          logc = self.env.log_cursor()
          logc.first()
          v = logc.next()
          self.assertNotEqual(v, logc.next())
          self.assertNotEqual(v, logc.next())
          self.assertEqual(v, logc.set(v[0]))
  
      def test_explicit_close(self) :
          logc = self.env.log_cursor()
          logc.close()
          self.assertRaises(db.DBCursorClosedError, logc.next)
  
      def test_implicit_close(self) :
          logc =  [self.env.log_cursor() for i in xrange(10)]
          self.env.close()  # This close should close too all its tree
          for i in logc :
              self.assertRaises(db.DBCursorClosedError, i.next)
  
  def test_suite():
      suite = unittest.TestSuite()
  
      suite.addTest(unittest.makeSuite(DBEnv_general))
      suite.addTest(unittest.makeSuite(DBEnv_memp))
      suite.addTest(unittest.makeSuite(DBEnv_logcursor))
      suite.addTest(unittest.makeSuite(DBEnv_log))
      suite.addTest(unittest.makeSuite(DBEnv_log_txn))
  
      return suite
  
  if __name__ == '__main__':
      unittest.main(defaultTest='test_suite')
  