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.
      """
  
  """
  Basic TestCases for BTree and hash DBs, with and without a DBEnv, with
  various DB flags, etc.
  """
  
  import os
  import errno
  import string
  from pprint import pprint
  import unittest
  import time
  import sys
  
  from .test_all import db, test_support, verbose, get_new_environment_path, \
          get_new_database_path
  
  DASH = '-'
  
  
  #----------------------------------------------------------------------
  
  class VersionTestCase(unittest.TestCase):
      def test00_version(self):
          info = db.version()
          if verbose:
              print('\n', '-=' * 20)
              print('bsddb.db.version(): %s' % (info, ))
              print(db.DB_VERSION_STRING)
              print('-=' * 20)
          self.assertEqual(info, (db.DB_VERSION_MAJOR, db.DB_VERSION_MINOR,
                          db.DB_VERSION_PATCH))
  
  #----------------------------------------------------------------------
  
  class BasicTestCase(unittest.TestCase):
      dbtype       = db.DB_UNKNOWN  # must be set in derived class
      cachesize    = (0, 1024*1024, 1)
      dbopenflags  = 0
      dbsetflags   = 0
      dbmode       = 0o660
      dbname       = None
      useEnv       = 0
      envflags     = 0
      envsetflags  = 0
  
      _numKeys      = 1002    # PRIVATE.  NOTE: must be an even value
  
      def setUp(self):
          if self.useEnv:
              self.homeDir=get_new_environment_path()
              try:
                  self.env = db.DBEnv()
                  self.env.set_lg_max(1024*1024)
                  self.env.set_tx_max(30)
                  self._t = int(time.time())
                  self.env.set_tx_timestamp(self._t)
                  self.env.set_flags(self.envsetflags, 1)
                  self.env.open(self.homeDir, self.envflags | db.DB_CREATE)
                  self.filename = "test"
              # Yes, a bare except is intended, since we're re-raising the exc.
              except:
                  test_support.rmtree(self.homeDir)
                  raise
          else:
              self.env = None
              self.filename = get_new_database_path()
  
          # create and open the DB
          self.d = db.DB(self.env)
          if not self.useEnv :
              self.d.set_cachesize(*self.cachesize)
              cachesize = self.d.get_cachesize()
              self.assertEqual(cachesize[0], self.cachesize[0])
              self.assertEqual(cachesize[2], self.cachesize[2])
              # Berkeley DB expands the cache 25% accounting overhead,
              # if the cache is small.
              self.assertEqual(125, int(100.0*cachesize[1]/self.cachesize[1]))
          self.d.set_flags(self.dbsetflags)
          if self.dbname:
              self.d.open(self.filename, self.dbname, self.dbtype,
                          self.dbopenflags|db.DB_CREATE, self.dbmode)
          else:
              self.d.open(self.filename,   # try out keyword args
                          mode = self.dbmode,
                          dbtype = self.dbtype,
                          flags = self.dbopenflags|db.DB_CREATE)
  
          if not self.useEnv:
              self.assertRaises(db.DBInvalidArgError,
                      self.d.set_cachesize, *self.cachesize)
  
          self.populateDB()
  
  
      def tearDown(self):
          self.d.close()
          if self.env is not None:
              self.env.close()
              test_support.rmtree(self.homeDir)
          else:
              os.remove(self.filename)
  
  
  
      def populateDB(self, _txn=None):
          d = self.d
  
          for x in range(self._numKeys//2):
              key = '%04d' % (self._numKeys - x)  # insert keys in reverse order
              data = self.makeData(key)
              d.put(key, data, _txn)
  
          d.put('empty value', '', _txn)
  
          for x in range(self._numKeys//2-1):
              key = '%04d' % x  # and now some in forward order
              data = self.makeData(key)
              d.put(key, data, _txn)
  
          if _txn:
              _txn.commit()
  
          num = len(d)
          if verbose:
              print("created %d records" % num)
  
  
      def makeData(self, key):
          return DASH.join([key] * 5)
  
  
  
      #----------------------------------------
  
      def test01_GetsAndPuts(self):
          d = self.d
  
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test01_GetsAndPuts..." % self.__class__.__name__)
  
          for key in ['0001', '0100', '0400', '0700', '0999']:
              data = d.get(key)
              if verbose:
                  print(data)
  
          self.assertEqual(d.get('0321'), '0321-0321-0321-0321-0321')
  
          # By default non-existent keys return None...
          self.assertEqual(d.get('abcd'), None)
  
          # ...but they raise exceptions in other situations.  Call
          # set_get_returns_none() to change it.
          try:
              d.delete('abcd')
          except db.DBNotFoundError as val:
              self.assertEqual(val.args[0], db.DB_NOTFOUND)
              if verbose: print(val)
          else:
              self.fail("expected exception")
  
  
          d.put('abcd', 'a new record')
          self.assertEqual(d.get('abcd'), 'a new record')
  
          d.put('abcd', 'same key')
          if self.dbsetflags & db.DB_DUP:
              self.assertEqual(d.get('abcd'), 'a new record')
          else:
              self.assertEqual(d.get('abcd'), 'same key')
  
  
          try:
              d.put('abcd', 'this should fail', flags=db.DB_NOOVERWRITE)
          except db.DBKeyExistError as val:
              self.assertEqual(val.args[0], db.DB_KEYEXIST)
              if verbose: print(val)
          else:
              self.fail("expected exception")
  
          if self.dbsetflags & db.DB_DUP:
              self.assertEqual(d.get('abcd'), 'a new record')
          else:
              self.assertEqual(d.get('abcd'), 'same key')
  
  
          d.sync()
          d.close()
          del d
  
          self.d = db.DB(self.env)
          if self.dbname:
              self.d.open(self.filename, self.dbname)
          else:
              self.d.open(self.filename)
          d = self.d
  
          self.assertEqual(d.get('0321'), '0321-0321-0321-0321-0321')
          if self.dbsetflags & db.DB_DUP:
              self.assertEqual(d.get('abcd'), 'a new record')
          else:
              self.assertEqual(d.get('abcd'), 'same key')
  
          rec = d.get_both('0555', '0555-0555-0555-0555-0555')
          if verbose:
              print(rec)
  
          self.assertEqual(d.get_both('0555', 'bad data'), None)
  
          # test default value
          data = d.get('bad key', 'bad data')
          self.assertEqual(data, 'bad data')
  
          # any object can pass through
          data = d.get('bad key', self)
          self.assertEqual(data, self)
  
          s = d.stat()
          self.assertEqual(type(s), type({}))
          if verbose:
              print('d.stat() returned this dictionary:')
              pprint(s)
  
  
      #----------------------------------------
  
      def test02_DictionaryMethods(self):
          d = self.d
  
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test02_DictionaryMethods..." % \
                    self.__class__.__name__)
  
          for key in ['0002', '0101', '0401', '0701', '0998']:
              data = d[key]
              self.assertEqual(data, self.makeData(key))
              if verbose:
                  print(data)
  
          self.assertEqual(len(d), self._numKeys)
          keys = list(d.keys())
          self.assertEqual(len(keys), self._numKeys)
          self.assertEqual(type(keys), type([]))
  
          d['new record'] = 'a new record'
          self.assertEqual(len(d), self._numKeys+1)
          keys = list(d.keys())
          self.assertEqual(len(keys), self._numKeys+1)
  
          d['new record'] = 'a replacement record'
          self.assertEqual(len(d), self._numKeys+1)
          keys = list(d.keys())
          self.assertEqual(len(keys), self._numKeys+1)
  
          if verbose:
              print("the first 10 keys are:")
              pprint(keys[:10])
  
          self.assertEqual(d['new record'], 'a replacement record')
  
  # We check also the positional parameter
          self.assertEqual(d.has_key('0001', None), 1)
  # We check also the keyword parameter
          self.assertEqual(d.has_key('spam', txn=None), 0)
  
          items = list(d.items())
          self.assertEqual(len(items), self._numKeys+1)
          self.assertEqual(type(items), type([]))
          self.assertEqual(type(items[0]), type(()))
          self.assertEqual(len(items[0]), 2)
  
          if verbose:
              print("the first 10 items are:")
              pprint(items[:10])
  
          values = list(d.values())
          self.assertEqual(len(values), self._numKeys+1)
          self.assertEqual(type(values), type([]))
  
          if verbose:
              print("the first 10 values are:")
              pprint(values[:10])
  
  
      #----------------------------------------
  
      def test02b_SequenceMethods(self):
          d = self.d
  
          for key in ['0002', '0101', '0401', '0701', '0998']:
              data = d[key]
              self.assertEqual(data, self.makeData(key))
              if verbose:
                  print(data)
  
          self.assertTrue(hasattr(d, "__contains__"))
          self.assertTrue("0401" in d)
          self.assertFalse("1234" in d)
  
  
      #----------------------------------------
  
      def test03_SimpleCursorStuff(self, get_raises_error=0, set_raises_error=0):
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test03_SimpleCursorStuff (get_error %s, set_error %s)..." % \
                    (self.__class__.__name__, get_raises_error, set_raises_error))
  
          if self.env and self.dbopenflags & db.DB_AUTO_COMMIT:
              txn = self.env.txn_begin()
          else:
              txn = None
          c = self.d.cursor(txn=txn)
  
          rec = c.first()
          count = 0
          while rec is not None:
              count = count + 1
              if verbose and count % 100 == 0:
                  print(rec)
              try:
                  rec = next(c)
              except db.DBNotFoundError as val:
                  if get_raises_error:
                      self.assertEqual(val.args[0], db.DB_NOTFOUND)
                      if verbose: print(val)
                      rec = None
                  else:
                      self.fail("unexpected DBNotFoundError")
              self.assertEqual(c.get_current_size(), len(c.current()[1]),
                      "%s != len(%r)" % (c.get_current_size(), c.current()[1]))
  
          self.assertEqual(count, self._numKeys)
  
  
          rec = c.last()
          count = 0
          while rec is not None:
              count = count + 1
              if verbose and count % 100 == 0:
                  print(rec)
              try:
                  rec = c.prev()
              except db.DBNotFoundError as val:
                  if get_raises_error:
                      self.assertEqual(val.args[0], db.DB_NOTFOUND)
                      if verbose: print(val)
                      rec = None
                  else:
                      self.fail("unexpected DBNotFoundError")
  
          self.assertEqual(count, self._numKeys)
  
          rec = c.set('0505')
          rec2 = c.current()
          self.assertEqual(rec, rec2)
          self.assertEqual(rec[0], '0505')
          self.assertEqual(rec[1], self.makeData('0505'))
          self.assertEqual(c.get_current_size(), len(rec[1]))
  
          # make sure we get empty values properly
          rec = c.set('empty value')
          self.assertEqual(rec[1], '')
          self.assertEqual(c.get_current_size(), 0)
  
          try:
              n = c.set('bad key')
          except db.DBNotFoundError as val:
              self.assertEqual(val.args[0], db.DB_NOTFOUND)
              if verbose: print(val)
          else:
              if set_raises_error:
                  self.fail("expected exception")
              if n is not None:
                  self.fail("expected None: %r" % (n,))
  
          rec = c.get_both('0404', self.makeData('0404'))
          self.assertEqual(rec, ('0404', self.makeData('0404')))
  
          try:
              n = c.get_both('0404', 'bad data')
          except db.DBNotFoundError as val:
              self.assertEqual(val.args[0], db.DB_NOTFOUND)
              if verbose: print(val)
          else:
              if get_raises_error:
                  self.fail("expected exception")
              if n is not None:
                  self.fail("expected None: %r" % (n,))
  
          if self.d.get_type() == db.DB_BTREE:
              rec = c.set_range('011')
              if verbose:
                  print("searched for '011', found: ", rec)
  
              rec = c.set_range('011',dlen=0,doff=0)
              if verbose:
                  print("searched (partial) for '011', found: ", rec)
              if rec[1] != '': self.fail('expected empty data portion')
  
              ev = c.set_range('empty value')
              if verbose:
                  print("search for 'empty value' returned", ev)
              if ev[1] != '': self.fail('empty value lookup failed')
  
          c.set('0499')
          c.delete()
          try:
              rec = c.current()
          except db.DBKeyEmptyError as val:
              if get_raises_error:
                  self.assertEqual(val.args[0], db.DB_KEYEMPTY)
                  if verbose: print(val)
              else:
                  self.fail("unexpected DBKeyEmptyError")
          else:
              if get_raises_error:
                  self.fail('DBKeyEmptyError exception expected')
  
          next(c)
          c2 = c.dup(db.DB_POSITION)
          self.assertEqual(c.current(), c2.current())
  
          c2.put('', 'a new value', db.DB_CURRENT)
          self.assertEqual(c.current(), c2.current())
          self.assertEqual(c.current()[1], 'a new value')
  
          c2.put('', 'er', db.DB_CURRENT, dlen=0, doff=5)
          self.assertEqual(c2.current()[1], 'a newer value')
  
          c.close()
          c2.close()
          if txn:
              txn.commit()
  
          # time to abuse the closed cursors and hope we don't crash
          methods_to_test = {
              'current': (),
              'delete': (),
              'dup': (db.DB_POSITION,),
              'first': (),
              'get': (0,),
              'next': (),
              'prev': (),
              'last': (),
              'put':('', 'spam', db.DB_CURRENT),
              'set': ("0505",),
          }
          for method, args in list(methods_to_test.items()):
              try:
                  if verbose:
                      print("attempting to use a closed cursor's %s method" % \
                            method)
                  # a bug may cause a NULL pointer dereference...
                  getattr(c, method)(*args)
              except db.DBError as val:
                  self.assertEqual(val.args[0], 0)
                  if verbose: print(val)
              else:
                  self.fail("no exception raised when using a buggy cursor's"
                            "%s method" % method)
  
          #
          # free cursor referencing a closed database, it should not barf:
          #
          oldcursor = self.d.cursor(txn=txn)
          self.d.close()
  
          # this would originally cause a segfault when the cursor for a
          # closed database was cleaned up.  it should not anymore.
          # SF pybsddb bug id 667343
          del oldcursor
  
      def test03b_SimpleCursorWithoutGetReturnsNone0(self):
          # same test but raise exceptions instead of returning None
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test03b_SimpleCursorStuffWithoutGetReturnsNone..." % \
                    self.__class__.__name__)
  
          old = self.d.set_get_returns_none(0)
          self.assertEqual(old, 2)
          self.test03_SimpleCursorStuff(get_raises_error=1, set_raises_error=1)
  
      def test03b_SimpleCursorWithGetReturnsNone1(self):
          # same test but raise exceptions instead of returning None
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test03b_SimpleCursorStuffWithoutGetReturnsNone..." % \
                    self.__class__.__name__)
  
          old = self.d.set_get_returns_none(1)
          self.test03_SimpleCursorStuff(get_raises_error=0, set_raises_error=1)
  
  
      def test03c_SimpleCursorGetReturnsNone2(self):
          # same test but raise exceptions instead of returning None
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test03c_SimpleCursorStuffWithoutSetReturnsNone..." % \
                    self.__class__.__name__)
  
          old = self.d.set_get_returns_none(1)
          self.assertEqual(old, 2)
          old = self.d.set_get_returns_none(2)
          self.assertEqual(old, 1)
          self.test03_SimpleCursorStuff(get_raises_error=0, set_raises_error=0)
  
      def test03d_SimpleCursorPriority(self) :
          c = self.d.cursor()
          c.set_priority(db.DB_PRIORITY_VERY_LOW)  # Positional
          self.assertEqual(db.DB_PRIORITY_VERY_LOW, c.get_priority())
          c.set_priority(priority=db.DB_PRIORITY_HIGH)  # Keyword
          self.assertEqual(db.DB_PRIORITY_HIGH, c.get_priority())
          c.close()
  
      #----------------------------------------
  
      def test04_PartialGetAndPut(self):
          d = self.d
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test04_PartialGetAndPut..." % \
                    self.__class__.__name__)
  
          key = "partialTest"
          data = "1" * 1000 + "2" * 1000
          d.put(key, data)
          self.assertEqual(d.get(key), data)
          self.assertEqual(d.get(key, dlen=20, doff=990),
                  ("1" * 10) + ("2" * 10))
  
          d.put("partialtest2", ("1" * 30000) + "robin" )
          self.assertEqual(d.get("partialtest2", dlen=5, doff=30000), "robin")
  
          # There seems to be a bug in DB here...  Commented out the test for
          # now.
          ##self.assertEqual(d.get("partialtest2", dlen=5, doff=30010), "")
  
          if self.dbsetflags != db.DB_DUP:
              # Partial put with duplicate records requires a cursor
              d.put(key, "0000", dlen=2000, doff=0)
              self.assertEqual(d.get(key), "0000")
  
              d.put(key, "1111", dlen=1, doff=2)
              self.assertEqual(d.get(key), "0011110")
  
      #----------------------------------------
  
      def test05_GetSize(self):
          d = self.d
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test05_GetSize..." % self.__class__.__name__)
  
          for i in range(1, 50000, 500):
              key = "size%s" % i
              #print "before ", i,
              d.put(key, "1" * i)
              #print "after",
              self.assertEqual(d.get_size(key), i)
              #print "done"
  
      #----------------------------------------
  
      def test06_Truncate(self):
          d = self.d
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test06_Truncate..." % self.__class__.__name__)
  
          d.put("abcde", "ABCDE");
          num = d.truncate()
          self.assertTrue(num >= 1, "truncate returned <= 0 on non-empty database")
          num = d.truncate()
          self.assertEqual(num, 0,
                  "truncate on empty DB returned nonzero (%r)" % (num,))
  
      #----------------------------------------
  
      def test07_verify(self):
          # Verify bug solved in 4.7.3pre8
          self.d.close()
          d = db.DB(self.env)
          d.verify(self.filename)
  
  
      #----------------------------------------
  
      def test08_exists(self) :
          self.d.put("abcde", "ABCDE")
          self.assertTrue(self.d.exists("abcde") == True,
                  "DB->exists() returns wrong value")
          self.assertTrue(self.d.exists("x") == False,
                  "DB->exists() returns wrong value")
  
      #----------------------------------------
  
      def test_compact(self) :
          d = self.d
          self.assertEqual(0, d.compact(flags=db.DB_FREELIST_ONLY))
          self.assertEqual(0, d.compact(flags=db.DB_FREELIST_ONLY))
          d.put("abcde", "ABCDE");
          d.put("bcde", "BCDE");
          d.put("abc", "ABC");
          d.put("monty", "python");
          d.delete("abc")
          d.delete("bcde")
          d.compact(start='abcde', stop='monty', txn=None,
                  compact_fillpercent=42, compact_pages=1,
                  compact_timeout=50000000,
                  flags=db.DB_FREELIST_ONLY|db.DB_FREE_SPACE)
  
      #----------------------------------------
  
  #----------------------------------------------------------------------
  
  
  class BasicBTreeTestCase(BasicTestCase):
      dbtype = db.DB_BTREE
  
  
  class BasicHashTestCase(BasicTestCase):
      dbtype = db.DB_HASH
  
  
  class BasicBTreeWithThreadFlagTestCase(BasicTestCase):
      dbtype = db.DB_BTREE
      dbopenflags = db.DB_THREAD
  
  
  class BasicHashWithThreadFlagTestCase(BasicTestCase):
      dbtype = db.DB_HASH
      dbopenflags = db.DB_THREAD
  
  
  class BasicWithEnvTestCase(BasicTestCase):
      dbopenflags = db.DB_THREAD
      useEnv = 1
      envflags = db.DB_THREAD | db.DB_INIT_MPOOL | db.DB_INIT_LOCK
  
      #----------------------------------------
  
      def test09_EnvRemoveAndRename(self):
          if not self.env:
              return
  
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test09_EnvRemoveAndRename..." % self.__class__.__name__)
  
          # can't rename or remove an open DB
          self.d.close()
  
          newname = self.filename + '.renamed'
          self.env.dbrename(self.filename, None, newname)
          self.env.dbremove(newname)
  
      #----------------------------------------
  
  class BasicBTreeWithEnvTestCase(BasicWithEnvTestCase):
      dbtype = db.DB_BTREE
  
  
  class BasicHashWithEnvTestCase(BasicWithEnvTestCase):
      dbtype = db.DB_HASH
  
  
  #----------------------------------------------------------------------
  
  class BasicTransactionTestCase(BasicTestCase):
      if sys.version_info < (2, 7) :
          def assertIn(self, a, b, msg=None) :
              return self.assertTrue(a in b, msg=msg)
  
      dbopenflags = db.DB_THREAD | db.DB_AUTO_COMMIT
      useEnv = 1
      envflags = (db.DB_THREAD | db.DB_INIT_MPOOL | db.DB_INIT_LOCK |
                  db.DB_INIT_TXN)
      envsetflags = db.DB_AUTO_COMMIT
  
  
      def tearDown(self):
          self.txn.commit()
          BasicTestCase.tearDown(self)
  
  
      def populateDB(self):
          txn = self.env.txn_begin()
          BasicTestCase.populateDB(self, _txn=txn)
  
          self.txn = self.env.txn_begin()
  
  
      def test06_Transactions(self):
          d = self.d
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test06_Transactions..." % self.__class__.__name__)
  
          self.assertEqual(d.get('new rec', txn=self.txn), None)
          d.put('new rec', 'this is a new record', self.txn)
          self.assertEqual(d.get('new rec', txn=self.txn),
                  'this is a new record')
          self.txn.abort()
          self.assertEqual(d.get('new rec'), None)
  
          self.txn = self.env.txn_begin()
  
          self.assertEqual(d.get('new rec', txn=self.txn), None)
          d.put('new rec', 'this is a new record', self.txn)
          self.assertEqual(d.get('new rec', txn=self.txn),
                  'this is a new record')
          self.txn.commit()
          self.assertEqual(d.get('new rec'), 'this is a new record')
  
          self.txn = self.env.txn_begin()
          c = d.cursor(self.txn)
          rec = c.first()
          count = 0
          while rec is not None:
              count = count + 1
              if verbose and count % 100 == 0:
                  print(rec)
              rec = next(c)
          self.assertEqual(count, self._numKeys+1)
  
          c.close()                # Cursors *MUST* be closed before commit!
          self.txn.commit()
  
          # flush pending updates
          self.env.txn_checkpoint (0, 0, 0)
  
          statDict = self.env.log_stat(0);
          self.assertIn('magic', statDict)
          self.assertIn('version', statDict)
          self.assertIn('cur_file', statDict)
          self.assertIn('region_nowait', statDict)
  
          # must have at least one log file present:
          logs = self.env.log_archive(db.DB_ARCH_ABS | db.DB_ARCH_LOG)
          self.assertNotEqual(logs, None)
          for log in logs:
              if verbose:
                  print('log file: ' + log)
          logs = self.env.log_archive(db.DB_ARCH_REMOVE)
          self.assertTrue(not logs)
  
          self.txn = self.env.txn_begin()
  
      #----------------------------------------
  
      def test08_exists(self) :
          txn = self.env.txn_begin()
          self.d.put("abcde", "ABCDE", txn=txn)
          txn.commit()
          txn = self.env.txn_begin()
          self.assertTrue(self.d.exists("abcde", txn=txn) == True,
                  "DB->exists() returns wrong value")
          self.assertTrue(self.d.exists("x", txn=txn) == False,
                  "DB->exists() returns wrong value")
          txn.abort()
  
      #----------------------------------------
  
      def test09_TxnTruncate(self):
          d = self.d
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test09_TxnTruncate..." % self.__class__.__name__)
  
          d.put("abcde", "ABCDE");
          txn = self.env.txn_begin()
          num = d.truncate(txn)
          self.assertTrue(num >= 1, "truncate returned <= 0 on non-empty database")
          num = d.truncate(txn)
          self.assertEqual(num, 0,
                  "truncate on empty DB returned nonzero (%r)" % (num,))
          txn.commit()
  
      #----------------------------------------
  
      def test10_TxnLateUse(self):
          txn = self.env.txn_begin()
          txn.abort()
          try:
              txn.abort()
          except db.DBError as e:
              pass
          else:
              raise RuntimeError("DBTxn.abort() called after DB_TXN no longer valid w/o an exception")
  
          txn = self.env.txn_begin()
          txn.commit()
          try:
              txn.commit()
          except db.DBError as e:
              pass
          else:
              raise RuntimeError("DBTxn.commit() called after DB_TXN no longer valid w/o an exception")
  
  
      #----------------------------------------
  
  
      def test_txn_name(self) :
          txn=self.env.txn_begin()
          self.assertEqual(txn.get_name(), "")
          txn.set_name("XXYY")
          self.assertEqual(txn.get_name(), "XXYY")
          txn.set_name("")
          self.assertEqual(txn.get_name(), "")
          txn.abort()
  
      #----------------------------------------
  
  
      def test_txn_set_timeout(self) :
          txn=self.env.txn_begin()
          txn.set_timeout(1234567, db.DB_SET_LOCK_TIMEOUT)
          txn.set_timeout(2345678, flags=db.DB_SET_TXN_TIMEOUT)
          txn.abort()
  
      #----------------------------------------
  
      def test_get_tx_max(self) :
          self.assertEqual(self.env.get_tx_max(), 30)
  
      def test_get_tx_timestamp(self) :
          self.assertEqual(self.env.get_tx_timestamp(), self._t)
  
  
  
  class BTreeTransactionTestCase(BasicTransactionTestCase):
      dbtype = db.DB_BTREE
  
  class HashTransactionTestCase(BasicTransactionTestCase):
      dbtype = db.DB_HASH
  
  
  
  #----------------------------------------------------------------------
  
  class BTreeRecnoTestCase(BasicTestCase):
      dbtype     = db.DB_BTREE
      dbsetflags = db.DB_RECNUM
  
      def test09_RecnoInBTree(self):
          d = self.d
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test09_RecnoInBTree..." % self.__class__.__name__)
  
          rec = d.get(200)
          self.assertEqual(type(rec), type(()))
          self.assertEqual(len(rec), 2)
          if verbose:
              print("Record #200 is ", rec)
  
          c = d.cursor()
          c.set('0200')
          num = c.get_recno()
          self.assertEqual(type(num), type(1))
          if verbose:
              print("recno of d['0200'] is ", num)
  
          rec = c.current()
          self.assertEqual(c.set_recno(num), rec)
  
          c.close()
  
  
  
  class BTreeRecnoWithThreadFlagTestCase(BTreeRecnoTestCase):
      dbopenflags = db.DB_THREAD
  
  #----------------------------------------------------------------------
  
  class BasicDUPTestCase(BasicTestCase):
      dbsetflags = db.DB_DUP
  
      def test10_DuplicateKeys(self):
          d = self.d
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test10_DuplicateKeys..." % \
                    self.__class__.__name__)
  
          d.put("dup0", "before")
          for x in "The quick brown fox jumped over the lazy dog.".split():
              d.put("dup1", x)
          d.put("dup2", "after")
  
          data = d.get("dup1")
          self.assertEqual(data, "The")
          if verbose:
              print(data)
  
          c = d.cursor()
          rec = c.set("dup1")
          self.assertEqual(rec, ('dup1', 'The'))
  
          next_reg = next(c)
          self.assertEqual(next_reg, ('dup1', 'quick'))
  
          rec = c.set("dup1")
          count = c.count()
          self.assertEqual(count, 9)
  
          next_dup = c.next_dup()
          self.assertEqual(next_dup, ('dup1', 'quick'))
  
          rec = c.set('dup1')
          while rec is not None:
              if verbose:
                  print(rec)
              rec = c.next_dup()
  
          c.set('dup1')
          rec = c.next_nodup()
          self.assertNotEqual(rec[0], 'dup1')
          if verbose:
              print(rec)
  
          c.close()
  
  
  
  class BTreeDUPTestCase(BasicDUPTestCase):
      dbtype = db.DB_BTREE
  
  class HashDUPTestCase(BasicDUPTestCase):
      dbtype = db.DB_HASH
  
  class BTreeDUPWithThreadTestCase(BasicDUPTestCase):
      dbtype = db.DB_BTREE
      dbopenflags = db.DB_THREAD
  
  class HashDUPWithThreadTestCase(BasicDUPTestCase):
      dbtype = db.DB_HASH
      dbopenflags = db.DB_THREAD
  
  
  #----------------------------------------------------------------------
  
  class BasicMultiDBTestCase(BasicTestCase):
      dbname = 'first'
  
      def otherType(self):
          if self.dbtype == db.DB_BTREE:
              return db.DB_HASH
          else:
              return db.DB_BTREE
  
      def test11_MultiDB(self):
          d1 = self.d
          if verbose:
              print('\n', '-=' * 30)
              print("Running %s.test11_MultiDB..." % self.__class__.__name__)
  
          d2 = db.DB(self.env)
          d2.open(self.filename, "second", self.dbtype,
                  self.dbopenflags|db.DB_CREATE)
          d3 = db.DB(self.env)
          d3.open(self.filename, "third", self.otherType(),
                  self.dbopenflags|db.DB_CREATE)
  
          for x in "The quick brown fox jumped over the lazy dog".split():
              d2.put(x, self.makeData(x))
  
          for x in string.ascii_letters:
              d3.put(x, x*70)
  
          d1.sync()
          d2.sync()
          d3.sync()
          d1.close()
          d2.close()
          d3.close()
  
          self.d = d1 = d2 = d3 = None
  
          self.d = d1 = db.DB(self.env)
          d1.open(self.filename, self.dbname, flags = self.dbopenflags)
          d2 = db.DB(self.env)
          d2.open(self.filename, "second",  flags = self.dbopenflags)
          d3 = db.DB(self.env)
          d3.open(self.filename, "third", flags = self.dbopenflags)
  
          c1 = d1.cursor()
          c2 = d2.cursor()
          c3 = d3.cursor()
  
          count = 0
          rec = c1.first()
          while rec is not None:
              count = count + 1
              if verbose and (count % 50) == 0:
                  print(rec)
              rec = next(c1)
          self.assertEqual(count, self._numKeys)
  
          count = 0
          rec = c2.first()
          while rec is not None:
              count = count + 1
              if verbose:
                  print(rec)
              rec = next(c2)
          self.assertEqual(count, 9)
  
          count = 0
          rec = c3.first()
          while rec is not None:
              count = count + 1
              if verbose:
                  print(rec)
              rec = next(c3)
          self.assertEqual(count, len(string.ascii_letters))
  
  
          c1.close()
          c2.close()
          c3.close()
  
          d2.close()
          d3.close()
  
  
  
  # Strange things happen if you try to use Multiple DBs per file without a
  # DBEnv with MPOOL and LOCKing...
  
  class BTreeMultiDBTestCase(BasicMultiDBTestCase):
      dbtype = db.DB_BTREE
      dbopenflags = db.DB_THREAD
      useEnv = 1
      envflags = db.DB_THREAD | db.DB_INIT_MPOOL | db.DB_INIT_LOCK
  
  class HashMultiDBTestCase(BasicMultiDBTestCase):
      dbtype = db.DB_HASH
      dbopenflags = db.DB_THREAD
      useEnv = 1
      envflags = db.DB_THREAD | db.DB_INIT_MPOOL | db.DB_INIT_LOCK
  
  
  class PrivateObject(unittest.TestCase) :
      def tearDown(self) :
          del self.obj
  
      def test01_DefaultIsNone(self) :
          self.assertEqual(self.obj.get_private(), None)
  
      def test02_assignment(self) :
          a = "example of private object"
          self.obj.set_private(a)
          b = self.obj.get_private()
          self.assertTrue(a is b)  # Object identity
  
      def test03_leak_assignment(self) :
          a = "example of private object"
          refcount = sys.getrefcount(a)
          self.obj.set_private(a)
          self.assertEqual(refcount+1, sys.getrefcount(a))
          self.obj.set_private(None)
          self.assertEqual(refcount, sys.getrefcount(a))
  
      def test04_leak_GC(self) :
          a = "example of private object"
          refcount = sys.getrefcount(a)
          self.obj.set_private(a)
          self.obj = None
          self.assertEqual(refcount, sys.getrefcount(a))
  
  class DBEnvPrivateObject(PrivateObject) :
      def setUp(self) :
          self.obj = db.DBEnv()
  
  class DBPrivateObject(PrivateObject) :
      def setUp(self) :
          self.obj = db.DB()
  
  class CrashAndBurn(unittest.TestCase) :
      #def test01_OpenCrash(self) :
      #    # See http://bugs.python.org/issue3307
      #    self.assertRaises(db.DBInvalidArgError, db.DB, None, 65535)
  
      if db.version() < (4, 8) :
          def test02_DBEnv_dealloc(self):
              # http://bugs.python.org/issue3885
              import gc
              self.assertRaises(db.DBInvalidArgError, db.DBEnv, ~db.DB_RPCCLIENT)
              gc.collect()
  
  
  #----------------------------------------------------------------------
  #----------------------------------------------------------------------
  
  def test_suite():
      suite = unittest.TestSuite()
  
      suite.addTest(unittest.makeSuite(VersionTestCase))
      suite.addTest(unittest.makeSuite(BasicBTreeTestCase))
      suite.addTest(unittest.makeSuite(BasicHashTestCase))
      suite.addTest(unittest.makeSuite(BasicBTreeWithThreadFlagTestCase))
      suite.addTest(unittest.makeSuite(BasicHashWithThreadFlagTestCase))
      suite.addTest(unittest.makeSuite(BasicBTreeWithEnvTestCase))
      suite.addTest(unittest.makeSuite(BasicHashWithEnvTestCase))
      suite.addTest(unittest.makeSuite(BTreeTransactionTestCase))
      suite.addTest(unittest.makeSuite(HashTransactionTestCase))
      suite.addTest(unittest.makeSuite(BTreeRecnoTestCase))
      suite.addTest(unittest.makeSuite(BTreeRecnoWithThreadFlagTestCase))
      suite.addTest(unittest.makeSuite(BTreeDUPTestCase))
      suite.addTest(unittest.makeSuite(HashDUPTestCase))
      suite.addTest(unittest.makeSuite(BTreeDUPWithThreadTestCase))
      suite.addTest(unittest.makeSuite(HashDUPWithThreadTestCase))
      suite.addTest(unittest.makeSuite(BTreeMultiDBTestCase))
      suite.addTest(unittest.makeSuite(HashMultiDBTestCase))
      suite.addTest(unittest.makeSuite(DBEnvPrivateObject))
      suite.addTest(unittest.makeSuite(DBPrivateObject))
      suite.addTest(unittest.makeSuite(CrashAndBurn))
  
      return suite
  
  
  if __name__ == '__main__':
      unittest.main(defaultTest='test_suite')
  