home *** CD-ROM | disk | FTP | other *** search
/ Maximum CD 2010 November / maximum-cd-2010-11.iso / DiscContents / calibre-0.7.13.msi / file_293 (.txt) < prev    next >
Encoding:
Python Compiled Bytecode  |  2010-08-06  |  6.3 KB  |  173 lines
  1. # Source Generated with Decompyle++
  2. # File: in.pyc (Python 2.6)
  3.  
  4. import os
  5. import unittest
  6. from test_all import db, test_support, verbose, get_new_environment_path, get_new_database_path
  7.  
  8. try:
  9.     import warnings
  10. except ImportError:
  11.     pass
  12.  
  13. warnings.filterwarnings('ignore', message = 'DB could not be closed in', category = RuntimeWarning)
  14.  
  15. class DBEnvClosedEarlyCrash(unittest.TestCase):
  16.     
  17.     def setUp(self):
  18.         self.homeDir = get_new_environment_path()
  19.         self.filename = 'test'
  20.  
  21.     
  22.     def tearDown(self):
  23.         test_support.rmtree(self.homeDir)
  24.  
  25.     
  26.     def test01_close_dbenv_before_db(self):
  27.         dbenv = db.DBEnv()
  28.         dbenv.open(self.homeDir, db.DB_INIT_CDB | db.DB_CREATE | db.DB_THREAD | db.DB_INIT_MPOOL, 438)
  29.         d = db.DB(dbenv)
  30.         d2 = db.DB(dbenv)
  31.         d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 438)
  32.         self.assertRaises(db.DBNoSuchFileError, d2.open, self.filename + '2', db.DB_BTREE, db.DB_THREAD, 438)
  33.         d.put('test', 'this is a test')
  34.         self.assertEqual(d.get('test'), 'this is a test', 'put!=get')
  35.         dbenv.close()
  36.         self.assertRaises(db.DBError, d.get, 'test')
  37.  
  38.     
  39.     def test02_close_dbenv_before_dbcursor(self):
  40.         dbenv = db.DBEnv()
  41.         dbenv.open(self.homeDir, db.DB_INIT_CDB | db.DB_CREATE | db.DB_THREAD | db.DB_INIT_MPOOL, 438)
  42.         d = db.DB(dbenv)
  43.         d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 438)
  44.         d.put('test', 'this is a test')
  45.         d.put('test2', 'another test')
  46.         d.put('test3', 'another one')
  47.         self.assertEqual(d.get('test'), 'this is a test', 'put!=get')
  48.         c = d.cursor()
  49.         c.first()
  50.         c.next()
  51.         d.close()
  52.         self.assertRaises(db.DBError, c.next)
  53.         d = db.DB(dbenv)
  54.         d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 438)
  55.         c = d.cursor()
  56.         c.first()
  57.         c.next()
  58.         dbenv.close()
  59.         self.assertRaises(db.DBError, c.next)
  60.  
  61.     
  62.     def test03_close_db_before_dbcursor_without_env(self):
  63.         import os.path as os
  64.         path = os.path.join(self.homeDir, self.filename)
  65.         d = db.DB()
  66.         d.open(path, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 438)
  67.         d.put('test', 'this is a test')
  68.         d.put('test2', 'another test')
  69.         d.put('test3', 'another one')
  70.         self.assertEqual(d.get('test'), 'this is a test', 'put!=get')
  71.         c = d.cursor()
  72.         c.first()
  73.         c.next()
  74.         d.close()
  75.         self.assertRaises(db.DBError, c.next)
  76.  
  77.     
  78.     def test04_close_massive(self):
  79.         dbenv = db.DBEnv()
  80.         dbenv.open(self.homeDir, db.DB_INIT_CDB | db.DB_CREATE | db.DB_THREAD | db.DB_INIT_MPOOL, 438)
  81.         dbs = [ db.DB(dbenv) for i in xrange(16) ]
  82.         cursors = []
  83.         for i in dbs:
  84.             i.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 438)
  85.         
  86.         dbs[10].put('test', 'this is a test')
  87.         dbs[10].put('test2', 'another test')
  88.         dbs[10].put('test3', 'another one')
  89.         self.assertEqual(dbs[4].get('test'), 'this is a test', 'put!=get')
  90.         for i in dbs:
  91.             []([ i.cursor() for j in xrange(32) ])
  92.         
  93.         for i in dbs[::3]:
  94.             i.close()
  95.         
  96.         for i in cursors[::3]:
  97.             i.close()
  98.         
  99.         self.assertRaises(db.DBError, dbs[9].get, 'test')
  100.         self.assertRaises(db.DBError, cursors[101].first)
  101.         cursors[80].first()
  102.         cursors[80].next()
  103.         dbenv.close()
  104.         self.assertRaises(db.DBError, cursors[80].next)
  105.  
  106.     
  107.     def test05_close_dbenv_delete_db_success(self):
  108.         dbenv = db.DBEnv()
  109.         dbenv.open(self.homeDir, db.DB_INIT_CDB | db.DB_CREATE | db.DB_THREAD | db.DB_INIT_MPOOL, 438)
  110.         d = db.DB(dbenv)
  111.         d.open(self.filename, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 438)
  112.         dbenv.close()
  113.         del d
  114.         
  115.         try:
  116.             import gc
  117.         except ImportError:
  118.             gc = None
  119.  
  120.         if gc:
  121.             gc.collect()
  122.         
  123.  
  124.     
  125.     def test06_close_txn_before_dup_cursor(self):
  126.         dbenv = db.DBEnv()
  127.         dbenv.open(self.homeDir, db.DB_INIT_TXN | db.DB_INIT_MPOOL | db.DB_INIT_LOG | db.DB_CREATE)
  128.         d = db.DB(dbenv)
  129.         txn = dbenv.txn_begin()
  130.         if db.version() < (4, 1):
  131.             d.open(self.filename, dbtype = db.DB_HASH, flags = db.DB_CREATE)
  132.         else:
  133.             d.open(self.filename, dbtype = db.DB_HASH, flags = db.DB_CREATE, txn = txn)
  134.         d.put('XXX', 'yyy', txn = txn)
  135.         txn.commit()
  136.         txn = dbenv.txn_begin()
  137.         c1 = d.cursor(txn)
  138.         c2 = c1.dup()
  139.         self.assertEquals(('XXX', 'yyy'), c1.first())
  140.         import warnings
  141.         warnings.catch_warnings().__enter__()
  142.         
  143.         try:
  144.             warnings.simplefilter('ignore')
  145.             txn.commit()
  146.         finally:
  147.             pass
  148.  
  149.         self.assertRaises(db.DBCursorClosedError, c2.first)
  150.  
  151.     if db.version() > (4, 3, 0):
  152.         
  153.         def test07_close_db_before_sequence(self):
  154.             import os.path as os
  155.             path = os.path.join(self.homeDir, self.filename)
  156.             d = db.DB()
  157.             d.open(path, db.DB_BTREE, db.DB_CREATE | db.DB_THREAD, 438)
  158.             dbs = db.DBSequence(d)
  159.             d.close()
  160.             dbs.close()
  161.  
  162.     
  163.  
  164.  
  165. def test_suite():
  166.     suite = unittest.TestSuite()
  167.     suite.addTest(unittest.makeSuite(DBEnvClosedEarlyCrash))
  168.     return suite
  169.  
  170. if __name__ == '__main__':
  171.     unittest.main(defaultTest = 'test_suite')
  172.  
  173.